From d403de1fdb784d7316b530f8dc47f57065ff07b1 Mon Sep 17 00:00:00 2001
From: ming luo <itestmycode@gmail.com>
Date: Thu, 28 Sep 2023 10:14:32 -0400
Subject: [PATCH] check batchBuilder in case batch is disabled

---
 pulsar/producer_partition.go | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 720c7df485..c2e880dcbe 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -867,6 +867,12 @@ type pendingItem struct {
 }
 
 func (p *partitionProducer) internalFlushCurrentBatch() {
+	if p.batchBuilder == nil {
+		// batch is not enabled
+		// the batch flush ticker should be stopped but it might still called once depends on when stop() is called concurrently
+		// so we add check to prevent the flow continues on a nil batchBuilder
+		return
+	}
 	if p.batchBuilder.IsMultiBatches() {
 		p.internalFlushCurrentBatches()
 		return