eolivelli commented on a change in pull request #12170:
URL: https://github.com/apache/pulsar/pull/12170#discussion_r731694720
##########
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:
catching `Throwable` is always hard to handle.
we should at least deal with `InterruptedException`, and in case of
InterruptedException you have to call `Thread.currentThread().interrupt() `
--
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]
