Shoothzj commented on a change in pull request #12170:
URL: https://github.com/apache/pulsar/pull/12170#discussion_r731714960
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/BatchMessageContainerImpl.java
##########
@@ -66,18 +66,28 @@ public boolean add(MessageImpl<?> msg, SendCallback
callback) {
}
if (++numMessagesInBatch == 1) {
- // some properties are common amongst the different messages in
the batch, hence we just pick it up from
- // the first message
- messageMetadata.setSequenceId(msg.getSequenceId());
- lowestSequenceId =
Commands.initBatchMessageMetadata(messageMetadata, msg.getMessageBuilder());
- this.firstCallback = callback;
- batchedMessageMetadataAndPayload = PulsarByteBufAllocator.DEFAULT
- .buffer(Math.min(maxBatchSize,
ClientCnx.getMaxMessageSize()));
- if (msg.getMessageBuilder().hasTxnidMostBits() &&
currentTxnidMostBits == -1) {
- currentTxnidMostBits =
msg.getMessageBuilder().getTxnidMostBits();
- }
- if (msg.getMessageBuilder().hasTxnidLeastBits() &&
currentTxnidLeastBits == -1) {
- currentTxnidLeastBits =
msg.getMessageBuilder().getTxnidLeastBits();
+ try {
+ // some properties are common amongst the different messages
in the batch, hence we just pick it up from
+ // the first message
+ messageMetadata.setSequenceId(msg.getSequenceId());
+ lowestSequenceId =
Commands.initBatchMessageMetadata(messageMetadata, msg.getMessageBuilder());
+ this.firstCallback = callback;
+ batchedMessageMetadataAndPayload =
PulsarByteBufAllocator.DEFAULT
+ .buffer(Math.min(maxBatchSize,
ClientCnx.getMaxMessageSize()));
+ if (msg.getMessageBuilder().hasTxnidMostBits() &&
currentTxnidMostBits == -1) {
+ currentTxnidMostBits =
msg.getMessageBuilder().getTxnidMostBits();
+ }
+ if (msg.getMessageBuilder().hasTxnidLeastBits() &&
currentTxnidLeastBits == -1) {
+ currentTxnidLeastBits =
msg.getMessageBuilder().getTxnidLeastBits();
+ }
+ } catch (Throwable e) {
Review comment:
This block can't throw `InterruptedException`.
We need to catch all exception here, because any exception can lead to a
uncover producer state.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
