data race when reconnecting to broker and resending pendingItems

[freeznet]

Race condition happened when reconnecting to broker and resending pendingItems. If pendingItem just timeout and complete by failTimeoutMessages, the pendingItem's buffer will be put back into bufferPool, and new batch builder will use the buffer, which will cause data race as the fault log shown below.

==================
WARNING: DATA RACE
Write at 0x00c00110ab20 by goroutine 348:
 runtime.slicecopy()
 /usr/local/go/src/runtime/slice.go:247 +0x0
 github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto.(*MessageMetadata).MarshalToSizedBuffer()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/pulsar_proto/PulsarApi.pb.go:7182 +0x9c4
 github.com/apache/pulsar-client-go/pulsar/internal.serializeBatch()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/commands.go:250 +0x329
 github.com/apache/pulsar-client-go/pulsar/internal.(*batchContainer).Flush()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/batch_builder.go:232 +0x544
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).internalFlushCurrentBatch()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:436 +0x75
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).internalSend()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:421 +0xc64
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).runEventsLoop()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:300 +0x21e
Previous read at 0x00c00110ab20 by goroutine 466:
 internal/race.ReadRange()
 /usr/local/go/src/internal/race/race.go:45 +0xb0
 syscall.Write()
 /usr/local/go/src/syscall/syscall_unix.go:215 +0x94
 internal/poll.ignoringEINTRIO()
 /usr/local/go/src/internal/poll/fd_unix.go:581 +0x16e
 internal/poll.(*FD).Write()
 /usr/local/go/src/internal/poll/fd_unix.go:274 +0x294
 net.(*netFD).Write()
 /usr/local/go/src/net/fd_posix.go:73 +0x68
 net.(*conn).Write()
 /usr/local/go/src/net/net.go:195 +0xeb
 net.(*TCPConn).Write()
 <autogenerated>:1 +0x69
 github.com/apache/pulsar-client-go/pulsar/internal.(*connection).internalWriteData()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/connection.go:449 +0x1f5
 github.com/apache/pulsar-client-go/pulsar/internal.(*connection).run()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/connection.go:391 +0x528
 github.com/apache/pulsar-client-go/pulsar/internal.(*connection).start.func1()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/connection.go:217 +0xc4
Goroutine 348 (running) created at:
 github.com/apache/pulsar-client-go/pulsar.newPartitionProducer()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:145 +0xe95
 github.com/apache/pulsar-client-go/pulsar.(*producer).internalCreatePartitionsProducers.func1()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_impl.go:210 +0xed
Goroutine 466 (finished) created at:
 github.com/apache/pulsar-client-go/pulsar/internal.(*connection).start()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/connection.go:213 +0x4c
 github.com/apache/pulsar-client-go/pulsar/internal.(*connectionPool).GetConnection()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/connection_pool.go:100 +0x8ec
 github.com/apache/pulsar-client-go/pulsar/internal.(*rpcClient).Request()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/internal/rpc_client.go:128 +0xed
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).grabCnx()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:192 +0xca6
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).reconnectToBroker()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:281 +0x25c
 github.com/apache/pulsar-client-go/pulsar.(*partitionProducer).runEventsLoop()
 /go/pkg/mod/github.com/apache/[email protected]/pulsar/producer_partition.go:308 +0x2c6
==================