data race in kafka sink

[amyangfei]

Bug Report

Please answer these questions before submitting your issue. Thanks!

1. What did you do? If possible, provide a recipe for reproducing the error.

Found data race in integration test

3. What did you see instead?

[2020-05-12T02:54:57.360Z] + echo /tmp/tidb_cdc_test/generate_column/stdout.log
[2020-05-12T02:54:57.360Z] /tmp/tidb_cdc_test/generate_column/stdout.log
[2020-05-12T02:54:57.360Z] + cat /tmp/tidb_cdc_test/generate_column/stdout.log
[2020-05-12T02:54:57.360Z] ==================
[2020-05-12T02:54:57.360Z] WARNING: DATA RACE
[2020-05-12T02:54:57.360Z] Write at 0x000005a33f70 by goroutine 136:
[2020-05-12T02:54:57.360Z]   github.com/pingcap/ticdc/cdc/sink/mqProducer.newSaramaConfig()
[2020-05-12T02:54:57.360Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/sink/mqProducer/kafka.go:258 +0x380
[2020-05-12T02:54:57.360Z]   github.com/pingcap/ticdc/cdc/sink/mqProducer.NewKafkaSaramaProducer()
[2020-05-12T02:54:57.360Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/sink/mqProducer/kafka.go:161 +0x27f
[2020-05-12T02:54:57.360Z]   github.com/pingcap/ticdc/cdc/sink.newKafkaSaramaSink()
[2020-05-12T02:54:57.360Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/sink/mq.go:302 +0x4af
[2020-05-12T02:54:57.360Z]   github.com/pingcap/ticdc/cdc/sink.NewSink()
[2020-05-12T02:54:57.360Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/sink/sink.go:82 +0x5c0
[2020-05-12T02:54:57.360Z]   github.com/pingcap/ticdc/cdc.runProcessor()
[2020-05-12T02:54:57.360Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/processor.go:830 +0x51e
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Capture).assignTask()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/capture.go:210 +0x58c
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Capture).handleTaskEvent()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/capture.go:182 +0x1c9
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Capture).Run()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/capture.go:140 +0x740
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Server).run.func2()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/server.go:203 +0xa5
[2020-05-12T02:54:57.361Z]   golang.org/x/sync/errgroup.(*Group).Go.func1()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/golang.org/x/sync@v0.0.0-20190911185100-cd5d95a43a6e/errgroup/errgroup.go:57 +0x71
[2020-05-12T02:54:57.361Z] 
[2020-05-12T02:54:57.361Z] Previous read at 0x000005a33f70 by goroutine 315:
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/encoder_decoder.go:29 +0x4b0
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*RecordBatch).encodeRecords()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/record_batch.go:200 +0xe5
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*RecordBatch).encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/record_batch.go:87 +0x714
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*Records).encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/records.go:63 +0xdc
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*ProduceRequest).encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/produce_request.go:106 +0x40c
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*request).encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/request.go:34 +0x1dc
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.encode()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/encoder_decoder.go:24 +0x12c
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*Broker).send()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/broker.go:712 +0x394
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*Broker).sendAndReceive()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/broker.go:741 +0x88
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*Broker).Produce()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/broker.go:349 +0x138
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*asyncProducer).newBrokerProducer.func1()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/async_producer.go:665 +0xdf
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.withRecover()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/utils.go:45 +0x5a
[2020-05-12T02:54:57.361Z] 
[2020-05-12T02:54:57.361Z] Goroutine 136 (running) created at:
[2020-05-12T02:54:57.361Z]   golang.org/x/sync/errgroup.(*Group).Go()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/golang.org/x/sync@v0.0.0-20190911185100-cd5d95a43a6e/errgroup/errgroup.go:54 +0x73
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Server).run()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/server.go:202 +0x542
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cdc.(*Server).Run()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cdc/server.go:137 +0x108
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cmd.runEServer()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/cmd/server.go:78 +0x3e2
[2020-05-12T02:54:57.361Z]   github.com/spf13/cobra.(*Command).execute()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/spf13/cobra@v0.0.5/command.go:826 +0x523
[2020-05-12T02:54:57.361Z]   github.com/spf13/cobra.(*Command).ExecuteC()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/spf13/cobra@v0.0.5/command.go:914 +0x41a
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc/cmd.Execute()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/spf13/cobra@v0.0.5/command.go:864 +0x66
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc.main()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/main.go:9 +0x47
[2020-05-12T02:54:57.361Z]   github.com/pingcap/ticdc.TestRunMain()
[2020-05-12T02:54:57.361Z]       /home/jenkins/agent/workspace/cdc_ghpr_kafka_integration_test/go/src/github.com/pingcap/ticdc/main_test.go:35 +0x15c
[2020-05-12T02:54:57.361Z]   testing.tRunner()
[2020-05-12T02:54:57.361Z]       /usr/local/go/src/testing/testing.go:909 +0x199
[2020-05-12T02:54:57.361Z] 
[2020-05-12T02:54:57.361Z] Goroutine 315 (running) created at:
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*asyncProducer).newBrokerProducer()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/async_producer.go:661 +0x4d1
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*asyncProducer).getBrokerProducer()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/async_producer.go:1081 +0x20f
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*partitionProducer).dispatch()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/async_producer.go:511 +0xffb
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.(*partitionProducer).dispatch-fm()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/async_producer.go:506 +0x41
[2020-05-12T02:54:57.361Z]   github.com/Shopify/sarama.withRecover()
[2020-05-12T02:54:57.361Z]       /go/pkg/mod/github.com/!shopify/sarama@v1.26.1/utils.go:45 +0x5a
[2020-05-12T02:54:57.361Z] ==================

4. Versions of the cluster

   • TiCDC version (execute cdc version):

     PR version: pingcap/ticdc@72ea55f