lhotari commented on code in PR #26135:
URL: https://github.com/apache/pulsar/pull/26135#discussion_r3519698434
##########
pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java:
##########
@@ -341,4 +351,49 @@ public void
testUpdateAutoScaleReceiverQueueHintRaceWithConcurrentDrain() {
"Hint must reflect the post-enqueue state (pipeline had >=1
message); "
+ "a concurrent drain of the just-enqueued message
must not clear it.");
}
+
+ @Test(invocationTimeOut = 1000)
+ public void testGetMessageAtSyncsAckSetInMessageIdWithBrokerAckSet() {
+ // Regression test for MessagePayloadContextImpl#getMessageAt: the
BatchMessageIdImpl handed
+ // back to the caller carries a shared ackSetInMessageId bitset that
must be seeded from the
+ // broker-reported ackSet, not a fresh "all unacked" bitset. Otherwise
indices the broker
+ // already knows are acked would be reported as still-outstanding in
the returned MessageId,
+ // which is the same root cause that let acked batch messages leak
into the DLQ (see
+ // ConsumerImpl#receiveIndividualMessagesFromBatch and its
ackSetInMessageId.and(...) fix).
+ final int batchSize = 3;
+ MessageMetadata messageMetadata = new MessageMetadata()
+ .setProducerName("test-producer")
+ .setSequenceId(0)
+ .setPublishTime(System.currentTimeMillis())
+ .setNumMessagesInBatch(batchSize);
+
+ // Broker reports index 0 as already acked (bit cleared); indices 1
and 2 are still
+ // outstanding (bits set). This mirrors the ackSet the broker attaches
on redelivery.
+ BitSet brokerAckSet = new BitSet(batchSize);
+ brokerAckSet.set(1);
+ brokerAckSet.set(2);
+ List<Long> ackSet =
Arrays.stream(brokerAckSet.toLongArray()).boxed().collect(Collectors.toList());
+
+ MessageIdImpl messageId = new MessageIdImpl(1L, 2L, -1);
+ MessagePayloadContextImpl context = MessagePayloadContextImpl.get(
+ null, messageMetadata, messageId, consumer, 0, ackSet,
DEFAULT_CONSUMER_EPOCH);
+ try {
+ // Index 0 is already acked per the broker, so it must not be
redelivered to the app.
+ MessagePayload payload0 =
MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{0}));
+ Assert.assertNull(context.getMessageAt(0, batchSize, payload0,
false, Schema.BYTES));
+
+ MessagePayload payload1 =
MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{1}));
+ Message<byte[]> message1 = context.getMessageAt(1, batchSize,
payload1, false, Schema.BYTES);
+ Assert.assertNotNull(message1);
Review Comment:
Could use `@Cleanup("release")` for both `MessagePayload` instances to avoid
the leak that @void-ptr974 pointed out.
--
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]
