[fix] [cli] the variable producerName of BatchMsgContainer is null (#…

…20819) Motivation: If the producer name is generated by the Broker, the producer will update the variable `producerName` after connecting, but not update the same variable of the batch message container. Modifications: fix bug (cherry picked from commit aba50f2)
apache · Jul 18, 2023 · 7ae3e8f · 7ae3e8f
1 parent 721275a
commit 7ae3e8f
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 10 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1598,11 +1598,12 @@ private void printSendCommandDebug(CommandSend send, ByteBuf headersAndPayload)
         headersAndPayload.resetReaderIndex();
         if (log.isDebugEnabled()) {
             log.debug("[{}] Received send message request. producer: {}:{} {}:{} size: {},"
-                            + " partition key is: {}, ordering key is {}",
+                            + " partition key is: {}, ordering key is {}, uncompressedSize is {}",
                     remoteAddress, send.getProducerId(), send.getSequenceId(), msgMetadata.getProducerName(),
                     msgMetadata.getSequenceId(), headersAndPayload.readableBytes(),
                     msgMetadata.hasPartitionKey() ? msgMetadata.getPartitionKey() : null,
-                    msgMetadata.hasOrderingKey() ? msgMetadata.getOrderingKey() : null);
+                    msgMetadata.hasOrderingKey() ? msgMetadata.getOrderingKey() : null,
+                    msgMetadata.getUncompressedSize());
         }
     }
 

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractBatchMessageContainer.java
@@ -35,7 +35,6 @@ public abstract class AbstractBatchMessageContainer implements BatchMessageConta
     protected CompressionType compressionType;
     protected CompressionCodec compressor;
     protected String topicName;
-    protected String producerName;
     protected ProducerImpl producer;
 
     protected int maxNumMessagesInBatch;
@@ -98,7 +97,6 @@ public ProducerImpl.OpSendMsg createOpSendMsg() throws IOException {
     public void setProducer(ProducerImpl<?> producer) {
         this.producer = producer;
         this.topicName = producer.getTopic();
-        this.producerName = producer.getProducerName();
         this.compressionType = CompressionCodecProvider
                 .convertToWireProtocol(producer.getConfiguration().getCompressionType());
         this.compressor = CompressionCodecProvider.getCompressionCodec(compressionType);

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
@@ -87,8 +87,8 @@ public BatchMessageContainerImpl(ProducerImpl<?> producer) {
     public boolean add(MessageImpl<?> msg, SendCallback callback) {
 
         if (log.isDebugEnabled()) {
-            log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName, producerName,
-                    numMessagesInBatch);
+            log.debug("[{}] [{}] add message to batch, num messages in batch so far {}", topicName,
+                    producer.getProducerName(), numMessagesInBatch);
         }
 
         if (++numMessagesInBatch == 1) {
@@ -205,8 +205,8 @@ public void discard(Exception ex) {
                 batchedMessageMetadataAndPayload = null;
             }
         } catch (Throwable t) {
-            log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topicName, producerName,
-                    lowestSequenceId, t);
+            log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topicName,
+                    producer.getProducerName(), lowestSequenceId, t);
         }
         clear();
     }
@@ -271,6 +271,14 @@ public OpSendMsg createOpSendMsg() throws IOException {
         ByteBufPair cmd = producer.sendMessage(producer.producerId, messageMetadata.getSequenceId(),
                 messageMetadata.getHighestSequenceId(), numMessagesInBatch, messageMetadata, encryptedPayload);
 
+        if (log.isDebugEnabled()) {
+            log.debug("[{}] [{}] Build batch msg seq:{}, highest-seq:{}, numMessagesInBatch: {}, uncompressedSize: {},"
+                            + " payloadSize: {}", topicName, producer.getProducerName(),
+                    messageMetadata.getSequenceId(), messageMetadata.getNumMessagesInBatch(),
+                    messageMetadata.getHighestSequenceId(),
+                    messageMetadata.getUncompressedSize(), encryptedPayload.readableBytes());
+        }
+
         OpSendMsg op = OpSendMsg.create(messages, cmd, messageMetadata.getSequenceId(),
                 messageMetadata.getHighestSequenceId(), firstCallback);
 

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageKeyBasedContainer.java
@@ -43,8 +43,8 @@ class BatchMessageKeyBasedContainer extends AbstractBatchMessageContainer {
     @Override
     public boolean add(MessageImpl<?> msg, SendCallback callback) {
         if (log.isDebugEnabled()) {
-            log.debug("[{}] [{}] add message to batch, num messages in batch so far is {}", topicName, producerName,
-                    numMessagesInBatch);
+            log.debug("[{}] [{}] add message to batch, num messages in batch so far is {}", topicName,
+                    producer.getProducerName(), numMessagesInBatch);
         }
         String key = getKey(msg);
         final BatchMessageContainerImpl batchMessageContainer = batches.computeIfAbsent(key,