Fix test

Author: RobertIndie

pulsar/producer_partition.go

@@ -1737,7 +1737,9 @@ func (i *pendingItem) done(err error) {
 
 	i.isDone = true
 	// return the buffer to the pool after all callbacks have been called.
-	defer i.buffer.Release()
+	if i.buffer != nil {
+		defer i.buffer.Release()
+	}
 	if i.flushCallback != nil {
 		i.flushCallback(err)
 	}

pulsar/producer_test.go

@@ -2383,7 +2383,9 @@ func TestFailPendingMessageWithClose(t *testing.T) {
 		})
 	}
 	partitionProducerImp := testProducer.(*producer).producers[0].(*partitionProducer)
-	partitionProducerImp.pendingQueue.Put(&pendingItem{})
+	partitionProducerImp.pendingQueue.Put(&pendingItem{
+		buffer: buffersPool.GetBuffer(0),
+	})
 	testProducer.Close()
 	assert.Equal(t, 0, partitionProducerImp.pendingQueue.Size())
 }