diff --git a/pkg/sink/kafka/sink.go b/pkg/sink/kafka/sink.go
index 660a0ed87..eca2b31e7 100644
--- a/pkg/sink/kafka/sink.go
+++ b/pkg/sink/kafka/sink.go
@@ -118,6 +118,11 @@ func (s *Sink) Start() error {
 func (s *Sink) Stop() {
 	if s.writer != nil {
 		_ = s.writer.Close()
+		if s.writer.Transport != nil {
+			if transport, ok := s.writer.Transport.(*kafka.Transport); ok {
+				transport.CloseIdleConnections()
+			}
+		}
 	}
 }