Merge pull request #98680 from HonoreDB/fix_kafka_external_connection

release-22.2: changefeedccl: fix kafka external connections
cockroachdb · Mar 15, 2023 · 5660a05 · 5660a05
2 parents 1f19637 + 45b38eb
commit 5660a05
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 4 deletions.
diff --git a/pkg/ccl/changefeedccl/sink.go b/pkg/ccl/changefeedccl/sink.go
@@ -212,8 +212,11 @@ func getSink(
 		sink = knobs.WrapSink(sink, jobID)
 	}
 
-	if err := sink.Dial(); err != nil {
-		return nil, err
+	// Avoid dialing twice. For external connections we've dialed inside the recursive call.
+	if u.Scheme != changefeedbase.SinkSchemeExternalConnection {
+		if err := sink.Dial(); err != nil {
+			return nil, err
+		}
 	}
 
 	return sink, nil

diff --git a/pkg/ccl/changefeedccl/sink_kafka.go b/pkg/ccl/changefeedccl/sink_kafka.go
@@ -303,8 +303,10 @@ func (s *kafkaSink) newSyncProducer(client kafkaClient) (sarama.SyncProducer, er
 
 // Close implements the Sink interface.
 func (s *kafkaSink) Close() error {
-	close(s.stopWorkerCh)
-	s.worker.Wait()
+	if s.stopWorkerCh != nil {
+		close(s.stopWorkerCh)
+		s.worker.Wait()
+	}
 
 	if s.producer != nil {
 		// Ignore errors related to outstanding messages since we're either shutting