fix: fix kafka sink panic (#3566)

Signed-off-by: Song Gao <disxiaofei@163.com>
lf-edge · Feb 24, 2025 · 6d4080c · 6d4080c
1 parent e067ff3
commit 6d4080c
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 0 deletions.
diff --git a/extensions/impl/kafka/sink.go b/extensions/impl/kafka/sink.go
@@ -304,6 +304,9 @@ func (k *KafkaSink) ingest(ctx api.StreamContext, d *kafkago.Message, checkSize
 }
 
 func (k *KafkaSink) send(ctx api.StreamContext) {
+	if len(k.messages) < 1 {
+		return
+	}
 	KafkaSinkCounter.WithLabelValues(LblSend, LblReq, k.ruleID, k.opID).Inc()
 	start := time.Now()
 	defer func() {

diff --git a/extensions/impl/kafka/sink_test.go b/extensions/impl/kafka/sink_test.go
@@ -180,3 +180,9 @@ func TestToCompression(t *testing.T) {
 		require.Equal(t, tc.expect, e)
 	}
 }
+
+func TestKafkaSinkEmpty(t *testing.T) {
+	ks := &KafkaSink{}
+	ctx := mockContext.NewMockContext("1", "2")
+	ks.send(ctx)
+}