Fix Kafka binding implements Close() for cleanup (#906)

daixiang0 · yaron2 · web-flow · commit d499387a44f3 · 2021-06-04T15:47:18.000-07:00
Signed-off-by: Long Dai &lt;long0dai@foxmail.com&gt;

Co-authored-by: Yaron Schneider &lt;yaronsc@microsoft.com&gt;
diff --git a/bindings/kafka/kafka.go b/bindings/kafka/kafka.go
@@ -273,3 +273,13 @@ func updateAuthInfo(config *sarama.Config, saslUsername, saslPassword string) {
 		ClientAuth: 0,
 	}
 }
+
+func (k *Kafka) Close() error {
+	if err := k.producer.Close(); err != nil {
+		k.logger.Errorf("kafka error: failed to close producer: %v", err)
+
+		return err
+	}
+
+	return nil
+}

Original file line number	Diff line number	Diff line change
`@@ -273,3 +273,13 @@ func updateAuthInfo(config *sarama.Config, saslUsername, saslPassword string) {`
`273`	`273`	`ClientAuth: 0,`
`274`	`274`	`}`
`275`	`275`	`}`
	`276`	`+`
	`277`	`+func (k *Kafka) Close() error {`
	`278`	`+ if err := k.producer.Close(); err != nil {`
	`279`	`+ k.logger.Errorf("kafka error: failed to close producer: %v", err)`
	`280`	`+`
	`281`	`+ return err`
	`282`	`+ }`
	`283`	`+`
	`284`	`+ return nil`
	`285`	`+}`