Skip to content

Commit ba52f1e

Browse files
BewareMyPowerBewareMyPower
committed
[fix][client] Fix deserialized BatchMessageIdImpl acknowledgment failure
Fixes apache#19030 ### Motivation When a `BatchMessageIdImpl` is created from a deserialization, the `BatchMessageAcker` object cannot be shared by all instances in the same batch, which leads to an acknowledgment failure when batch index ACK is disabled (by default). ### Modifications Maintain a map from the `(ledger id, entry id)` pair to the `BatchMessageAcker` in `ConsumerImpl`. If the `BatchMessageIdImpl` doesn't carry a valid `BatchMessageAcker`, create and cache a `BatchMessageAcker` instance and remove it when all messages in the batch are acknowledged. It requires a change in `MessageIdImpl#fromByteArray` that a `BatchMessageAckerDisabled` will be created to indicate there is no shared acker. To avoid making code more complicated, this patch refactors the existing code that many logics about consumer are moved from the ACK tracker to the consumer. It also removes the `AckType` parameter when acknowledging a list of messages.
1 parent 41edd2e commit ba52f1e

9 files changed

Lines changed: 261 additions & 187 deletions

File tree

‎pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageIdSerializationTest.java‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.pulsar.client.api;
20+
21+
import static org.testng.Assert.assertEquals;
22+
import static org.testng.Assert.assertTrue;
23+
import java.util.ArrayList;
24+
import java.util.List;
25+
import java.util.concurrent.TimeUnit;
26+
import java.util.concurrent.atomic.AtomicLong;
27+
import lombok.Cleanup;
28+
import org.apache.pulsar.client.impl.BatchMessageIdImpl;
29+
import org.testng.annotations.AfterClass;
30+
import org.testng.annotations.BeforeClass;
31+
import org.testng.annotations.Test;
32+
33+
@Test(groups = "broker-api")
34+
public class MessageIdSerializationTest extends ProducerConsumerBase {
35+
36+
@BeforeClass
37+
@Override
38+
protected void setup() throws Exception {
39+
super.internalSetup();
40+
super.producerBaseSetup();
41+
}
42+
43+
@AfterClass(alwaysRun = true)
44+
@Override
45+
protected void cleanup() throws Exception {
46+
super.internalCleanup();
47+
}
48+
49+
@Test(timeOut = 30000)
50+
public void testSerialization() throws Exception {
51+
String topic = "test-serialization-origin";
52+
@Cleanup Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
53+
.topic(topic)
54+
.batchingMaxMessages(100)
55+
.batchingMaxPublishDelay(1, TimeUnit.DAYS)
56+
.create();
57+
Consumer<Integer> consumer = pulsarClient.newConsumer(Schema.INT32)
58+
.topic(topic)
59+
.subscriptionName("sub")
60+
.isAckReceiptEnabled(true)
61+
.subscribe();
62+
63+
final int numMessages = 10;
64+
for (int i = 0; i < numMessages; i++) {
65+
producer.sendAsync(i);
66+
}
67+
producer.flush();
68+
final List<MessageId> msgIds = new ArrayList<>();
69+
for (int i = 0; i < numMessages; i++) {
70+
msgIds.add(consumer.receive().getMessageId());
71+
}
72+
final AtomicLong ledgerId = new AtomicLong(-1L);
73+
final AtomicLong entryId = new AtomicLong(-1L);
74+
for (int i = 0; i < numMessages; i++) {
75+
assertTrue(msgIds.get(i) instanceof BatchMessageIdImpl);
76+
final BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) msgIds.get(i);
77+
ledgerId.compareAndSet(-1L, batchMessageId.getLedgerId());
78+
assertEquals(batchMessageId.getLedgerId(), ledgerId.get());
79+
entryId.compareAndSet(-1L, batchMessageId.getEntryId());
80+
assertEquals(batchMessageId.getEntryId(), entryId.get());
81+
assertEquals(batchMessageId.getBatchSize(), numMessages);
82+
}
83+
84+
final List<MessageId> deserializedMsgIds = new ArrayList<>();
85+
for (MessageId msgId : msgIds) {
86+
MessageId deserialized = MessageId.fromByteArray(msgId.toByteArray());
87+
assertTrue(deserialized instanceof BatchMessageIdImpl);
88+
deserializedMsgIds.add(deserialized);
89+
}
90+
for (MessageId msgId : deserializedMsgIds) {
91+
consumer.acknowledge(msgId);
92+
}
93+
consumer.close();
94+
95+
consumer = pulsarClient.newConsumer(Schema.INT32)
96+
.topic(topic)
97+
.subscriptionName("sub")
98+
.isAckReceiptEnabled(true)
99+
.subscribe();
100+
MessageId newMsgId = producer.send(0);
101+
MessageId receivedMessageId = consumer.receive().getMessageId();
102+
assertEquals(newMsgId, receivedMessageId);
103+
consumer.close();
104+
}
105+
}

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/AcknowledgmentsGroupingTracker.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,12 @@ public interface AcknowledgmentsGroupingTracker extends AutoCloseable {
3333

3434
CompletableFuture<Void> addAcknowledgment(MessageIdImpl msgId, AckType ackType, Map<String, Long> properties);
3535

36-
CompletableFuture<Void> addListAcknowledgment(List<MessageId> messageIds, AckType ackType,
37-
Map<String, Long> properties);
36+
default CompletableFuture<Void> addBatchIndexAck(BatchMessageIdImpl msgId, AckType ackType,
37+
Map<String, Long> properties) {
38+
return CompletableFuture.completedFuture(null);
39+
}
40+
41+
CompletableFuture<Void> addListAcknowledgment(List<MessageIdImpl> messageIds, Map<String, Long> properties);
3842

3943
void flush();
4044

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -542,12 +542,12 @@ public CompletableFuture<Void> acknowledgeAsync(Messages<?> messages, Transactio
542542

543543
@Override
544544
public CompletableFuture<Void> acknowledgeAsync(List<MessageId> messageIdList) {
545-
return doAcknowledgeWithTxn(messageIdList, AckType.Individual, Collections.emptyMap(), null);
545+
return doAcknowledgeWithTxn(messageIdList, Collections.emptyMap(), null);
546546
}
547547

548548
@Override
549549
public CompletableFuture<Void> acknowledgeAsync(List<MessageId> messageIdList, Transaction txn) {
550-
return doAcknowledgeWithTxn(messageIdList, AckType.Individual, Collections.emptyMap(), (TransactionImpl) txn);
550+
return doAcknowledgeWithTxn(messageIdList, Collections.emptyMap(), (TransactionImpl) txn);
551551
}
552552

553553
@Override
@@ -655,17 +655,17 @@ public void negativeAcknowledge(Message<?> message) {
655655
negativeAcknowledge(message.getMessageId());
656656
}
657657

658-
protected CompletableFuture<Void> doAcknowledgeWithTxn(List<MessageId> messageIdList, AckType ackType,
658+
protected CompletableFuture<Void> doAcknowledgeWithTxn(List<MessageId> messageIdList,
659659
Map<String, Long> properties,
660660
TransactionImpl txn) {
661661
CompletableFuture<Void> ackFuture;
662662
if (txn != null && this instanceof ConsumerImpl) {
663663
ackFuture = txn.registerAckedTopic(getTopic(), subscription)
664-
.thenCompose(ignored -> doAcknowledge(messageIdList, ackType, properties, txn));
664+
.thenCompose(ignored -> doAcknowledge(messageIdList, AckType.Individual, properties, txn));
665665
// register the ackFuture as part of the transaction
666666
txn.registerAckOp(ackFuture);
667667
} else {
668-
ackFuture = doAcknowledge(messageIdList, ackType, properties, txn);
668+
ackFuture = doAcknowledge(messageIdList, AckType.Individual, properties, txn);
669669
}
670670
return ackFuture;
671671
}

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java‎

Lines changed: 92 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,11 @@
6262
import java.util.concurrent.locks.ReentrantReadWriteLock;
6363
import java.util.function.Function;
6464
import java.util.stream.Collectors;
65+
import javax.annotation.Nullable;
6566
import lombok.AccessLevel;
6667
import lombok.Getter;
6768
import org.apache.commons.lang3.StringUtils;
69+
import org.apache.commons.lang3.tuple.Pair;
6870
import org.apache.pulsar.client.api.Consumer;
6971
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
7072
import org.apache.pulsar.client.api.DeadLetterPolicy;
@@ -204,6 +206,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
204206

205207
private final AtomicReference<ClientCnx> clientCnxUsedForConsumerRegistration = new AtomicReference<>();
206208
private final List<Throwable> previousExceptions = new CopyOnWriteArrayList<Throwable>();
209+
// Key is the ledger id and the entry id, entry is the acker that represents which single messages are acknowledged
210+
private final Map<Pair<Long, Long>, BatchMessageAcker> batchMessageToAcker = new ConcurrentHashMap<>();
211+
207212
static <T> ConsumerImpl<T> newConsumerImpl(PulsarClientImpl client,
208213
String topic,
209214
ConsumerConfigurationData<T> conf,
@@ -529,6 +534,52 @@ protected CompletableFuture<Messages<T>> internalBatchReceiveAsync() {
529534
return result;
530535
}
531536

537+
private void processMessageIdBeforeAcknowledge(MessageIdImpl messageId, AckType ackType, int numMessages) {
538+
if (ackType == AckType.Individual) {
539+
stats.incrementNumAcksSent(numMessages);
540+
unAckedMessageTracker.remove(messageId);
541+
if (possibleSendToDeadLetterTopicMessages != null) {
542+
possibleSendToDeadLetterTopicMessages.remove(messageId);
543+
}
544+
} else {
545+
stats.incrementNumAcksSent(unAckedMessageTracker.removeMessagesTill(messageId));
546+
}
547+
}
548+
549+
private @Nullable MessageIdImpl getMessageIdToAcknowledge(BatchMessageIdImpl messageId, AckType ackType) {
550+
if (conf.isBatchIndexAckEnabled()) {
551+
return messageId;
552+
}
553+
final BatchMessageAcker acker;
554+
if (messageId.getAcker() instanceof BatchMessageAckerDisabled) {
555+
acker = batchMessageToAcker.computeIfAbsent(
556+
Pair.of(messageId.getLedgerId(), messageId.getEntryId()),
557+
__ -> BatchMessageAcker.newAcker(messageId.getOriginalBatchSize()));
558+
} else {
559+
acker = messageId.getAcker();
560+
}
561+
if (ackType == AckType.Individual) {
562+
if (acker.ackIndividual(messageId.getBatchIndex())) {
563+
batchMessageToAcker.remove(Pair.of(messageId.getLedgerId(), messageId.getEntryId()));
564+
return messageId.toMessageIdImpl();
565+
} else {
566+
return null;
567+
}
568+
} else {
569+
if (acker.ackCumulative(messageId.getBatchIndex())) {
570+
batchMessageToAcker.remove(Pair.of(messageId.getLedgerId(), messageId.getEntryId()));
571+
return messageId.toMessageIdImpl();
572+
} else {
573+
if (acker.isPrevBatchCumulativelyAcked()) {
574+
return null;
575+
} else {
576+
acker.setPrevBatchCumulativelyAcked(true);
577+
return messageId.prevBatchMessageId();
578+
}
579+
}
580+
}
581+
}
582+
532583
@Override
533584
protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType,
534585
Map<String, Long> properties,
@@ -549,13 +600,34 @@ protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ack
549600
return doTransactionAcknowledgeForResponse(messageId, ackType, null, properties,
550601
new TxnID(txn.getTxnIdMostBits(), txn.getTxnIdLeastBits()));
551602
}
552-
return acknowledgmentsGroupingTracker.addAcknowledgment((MessageIdImpl) messageId, ackType, properties);
603+
if (ackType == AckType.Individual) {
604+
onAcknowledge(messageId, null);
605+
} else {
606+
onAcknowledgeCumulative(messageId, null);
607+
}
608+
if (messageId instanceof BatchMessageIdImpl) {
609+
BatchMessageIdImpl batchMessageId = (BatchMessageIdImpl) messageId;
610+
MessageIdImpl messageIdImpl = getMessageIdToAcknowledge(batchMessageId, ackType);
611+
if (messageIdImpl == null) {
612+
return CompletableFuture.completedFuture(null);
613+
} else if (messageIdImpl instanceof BatchMessageIdImpl) {
614+
return acknowledgmentsGroupingTracker.addBatchIndexAck(
615+
(BatchMessageIdImpl) messageIdImpl, ackType, properties);
616+
} else {
617+
processMessageIdBeforeAcknowledge(messageIdImpl, ackType, batchMessageId.getOriginalBatchSize());
618+
return acknowledgmentsGroupingTracker.addAcknowledgment(messageIdImpl, ackType, properties);
619+
}
620+
} else {
621+
MessageIdImpl messageIdImpl = (MessageIdImpl) messageId;
622+
processMessageIdBeforeAcknowledge(messageIdImpl, ackType, 1);
623+
return acknowledgmentsGroupingTracker.addAcknowledgment(messageIdImpl, ackType, properties);
624+
}
553625
}
554626

555627
@Override
556628
protected CompletableFuture<Void> doAcknowledge(List<MessageId> messageIdList, AckType ackType,
557629
Map<String, Long> properties, TransactionImpl txn) {
558-
630+
List<MessageIdImpl> messageIdListToAck = new ArrayList<>();
559631
for (MessageId messageId : messageIdList) {
560632
checkArgument(messageId instanceof MessageIdImpl);
561633
}
@@ -573,7 +645,24 @@ protected CompletableFuture<Void> doAcknowledge(List<MessageId> messageIdList, A
573645
return doTransactionAcknowledgeForResponse(messageIdList, ackType, null,
574646
properties, new TxnID(txn.getTxnIdMostBits(), txn.getTxnIdLeastBits()));
575647
} else {
576-
return this.acknowledgmentsGroupingTracker.addListAcknowledgment(messageIdList, ackType, properties);
648+
for (MessageId messageId : messageIdList) {
649+
checkArgument(messageId instanceof MessageIdImpl);
650+
onAcknowledge(messageId, null);
651+
if (messageId instanceof BatchMessageIdImpl) {
652+
MessageIdImpl messageIdImpl = getMessageIdToAcknowledge((BatchMessageIdImpl) messageId, ackType);
653+
if (messageIdImpl != null) {
654+
if (!(messageIdImpl instanceof BatchMessageIdImpl)) {
655+
processMessageIdBeforeAcknowledge(messageIdImpl, ackType, 1);
656+
}
657+
messageIdListToAck.add(messageIdImpl);
658+
}
659+
} else {
660+
MessageIdImpl messageIdImpl = (MessageIdImpl) messageId;
661+
processMessageIdBeforeAcknowledge(messageIdImpl, ackType, 1);
662+
messageIdListToAck.add(messageIdImpl);
663+
}
664+
}
665+
return this.acknowledgmentsGroupingTracker.addListAcknowledgment(messageIdListToAck, properties);
577666
}
578667
}
579668

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdImpl.java‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -95,14 +95,9 @@ public static MessageId fromByteArray(byte[] data) throws IOException {
9595
}
9696

9797
MessageIdImpl messageId;
98-
if (idData.hasBatchIndex()) {
99-
if (idData.hasBatchSize()) {
100-
messageId = new BatchMessageIdImpl(idData.getLedgerId(), idData.getEntryId(), idData.getPartition(),
101-
idData.getBatchIndex(), idData.getBatchSize(), BatchMessageAcker.newAcker(idData.getBatchSize()));
102-
} else {
103-
messageId = new BatchMessageIdImpl(idData.getLedgerId(), idData.getEntryId(), idData.getPartition(),
104-
idData.getBatchIndex());
105-
}
98+
if (idData.hasBatchIndex() && idData.hasBatchSize()) {
99+
messageId = new BatchMessageIdImpl(idData.getLedgerId(), idData.getEntryId(), idData.getPartition(),
100+
idData.getBatchIndex(), idData.getBatchSize(), BatchMessageAckerDisabled.INSTANCE);
106101
} else if (idData.hasFirstChunkMessageId()) {
107102
MessageIdData firstChunkIdData = idData.getFirstChunkMessageId();
108103
messageId = new ChunkMessageIdImpl(

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -493,7 +493,7 @@ protected CompletableFuture<Void> doAcknowledge(List<MessageId> messageIdList,
493493
}
494494
topicToMessageIdMap.forEach((topicPartitionName, messageIds) -> {
495495
ConsumerImpl<T> consumer = consumers.get(topicPartitionName);
496-
resultFutures.add(consumer.doAcknowledgeWithTxn(messageIds, ackType, properties, txn)
496+
resultFutures.add(consumer.doAcknowledgeWithTxn(messageIds, properties, txn)
497497
.thenAccept((res) -> messageIdList.forEach(unAckedMessageTracker::remove)));
498498
});
499499
return CompletableFuture.allOf(resultFutures.toArray(new CompletableFuture[0]));

‎pulsar-client/src/main/java/org/apache/pulsar/client/impl/NonPersistentAcknowledgmentGroupingTracker.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,7 @@ public CompletableFuture<Void> addAcknowledgment(MessageIdImpl msgId, AckType ac
5050
}
5151

5252
@Override
53-
public CompletableFuture<Void> addListAcknowledgment(List<MessageId> messageIds,
54-
AckType ackType,
53+
public CompletableFuture<Void> addListAcknowledgment(List<MessageIdImpl> messageIds,
5554
Map<String, Long> properties) {
5655
// no-op
5756
return CompletableFuture.completedFuture(null);

0 commit comments

Comments
 (0)