You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/03/27 12:45:43 UTC
(pulsar) 04/04: [fix][client] Fix wrong results of hasMessageAvailable and readNext after seeking by timestamp (#22363)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1c886bade09cd801ca809eb0c453dcaf0126ae93
Author: Yunze Xu <xy...@163.com>
AuthorDate: Wed Mar 27 19:49:27 2024 +0800
[fix][client] Fix wrong results of hasMessageAvailable and readNext after seeking by timestamp (#22363)
Co-authored-by: Lari Hotari <lh...@users.noreply.github.com>
(cherry picked from commit 149deaa5a79ed8570489bead4215ae213a4e9206)
---
.../org/apache/pulsar/client/impl/ReaderTest.java | 84 +++++++++++++++++++---
.../apache/pulsar/client/impl/ConsumerImpl.java | 25 ++++---
2 files changed, 89 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index cee3ea09968..2d3e8d4c6e9 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -66,8 +66,8 @@ import org.apache.pulsar.common.util.Murmur3_32Hash;
import org.apache.pulsar.schema.Schemas;
import org.awaitility.Awaitility;
import org.testng.Assert;
-import org.testng.annotations.AfterMethod;
-import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -77,7 +77,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
private static final String subscription = "reader-sub";
- @BeforeMethod
+ @BeforeClass(alwaysRun = true)
@Override
protected void setup() throws Exception {
super.internalSetup();
@@ -89,7 +89,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
admin.namespaces().createNamespace("my-property/my-ns", Sets.newHashSet("test"));
}
- @AfterMethod(alwaysRun = true)
+ @AfterClass(alwaysRun = true)
@Override
protected void cleanup() throws Exception {
super.internalCleanup();
@@ -198,21 +198,41 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
testReadMessages(topic, true);
}
- @Test
- public void testReadMessageWithBatchingWithMessageInclusive() throws Exception {
+ @DataProvider
+ public static Object[][] seekBeforeHasMessageAvailable() {
+ return new Object[][] { { true }, { false } };
+ }
+
+ @Test(timeOut = 20000, dataProvider = "seekBeforeHasMessageAvailable")
+ public void testReadMessageWithBatchingWithMessageInclusive(boolean seekBeforeHasMessageAvailable)
+ throws Exception {
String topic = "persistent://my-property/my-ns/my-reader-topic-with-batching-inclusive";
Set<String> keys = publishMessages(topic, 10, true);
Reader<byte[]> reader = pulsarClient.newReader().topic(topic).startMessageId(MessageId.latest)
.startMessageIdInclusive().readerName(subscription).create();
- while (reader.hasMessageAvailable()) {
- Assert.assertTrue(keys.remove(reader.readNext().getKey()));
+ if (seekBeforeHasMessageAvailable) {
+ reader.seek(0L); // it should seek to the earliest
}
+
+ assertTrue(reader.hasMessageAvailable());
+ final Message<byte[]> msg = reader.readNext();
+ assertTrue(keys.remove(msg.getKey()));
// start from latest with start message inclusive should only read the last message in batch
assertEquals(keys.size(), 9);
- Assert.assertFalse(keys.contains("key9"));
- Assert.assertFalse(reader.hasMessageAvailable());
+
+ final MessageIdAdv msgId = (MessageIdAdv) msg.getMessageId();
+ if (seekBeforeHasMessageAvailable) {
+ assertEquals(msgId.getBatchIndex(), 0);
+ assertFalse(keys.contains("key0"));
+ assertTrue(reader.hasMessageAvailable());
+ } else {
+ assertEquals(msgId.getBatchIndex(), 9);
+ assertFalse(reader.hasMessageAvailable());
+ assertFalse(keys.contains("key9"));
+ assertFalse(reader.hasMessageAvailable());
+ }
}
private void testReadMessages(String topic, boolean enableBatch) throws Exception {
@@ -310,7 +330,7 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
@Test
public void testReaderWithTimeLong() throws Exception {
String ns = "my-property/my-ns";
- String topic = "persistent://" + ns + "/testReadFromPartition";
+ String topic = "persistent://" + ns + "/testReaderWithTimeLong";
RetentionPolicies retention = new RetentionPolicies(-1, -1);
admin.namespaces().setRetention(ns, retention);
@@ -840,4 +860,46 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
producer.send("msg");
assertTrue(reader.hasMessageAvailable());
}
+
+ @Test(dataProvider = "initializeLastMessageIdInBroker")
+ public void testHasMessageAvailableAfterSeekTimestamp(boolean initializeLastMessageIdInBroker) throws Exception {
+ final String topic = "persistent://my-property/my-ns/test-has-message-available-after-seek-timestamp";
+
+ @Cleanup Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).create();
+ final long timestampBeforeSend = System.currentTimeMillis();
+ final MessageId sentMsgId = producer.send("msg");
+
+ final List<MessageId> messageIds = new ArrayList<>();
+ messageIds.add(MessageId.earliest);
+ messageIds.add(sentMsgId);
+ messageIds.add(MessageId.latest);
+
+ for (MessageId messageId : messageIds) {
+ @Cleanup Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
+ .startMessageId(messageId).create();
+ if (initializeLastMessageIdInBroker) {
+ if (messageId == MessageId.earliest) {
+ assertTrue(reader.hasMessageAvailable());
+ } else {
+ assertFalse(reader.hasMessageAvailable());
+ }
+ } // else: lastMessageIdInBroker is earliest
+ reader.seek(System.currentTimeMillis());
+ assertFalse(reader.hasMessageAvailable());
+ }
+
+ for (MessageId messageId : messageIds) {
+ @Cleanup Reader<String> reader = pulsarClient.newReader(Schema.STRING).topic(topic).receiverQueueSize(1)
+ .startMessageId(messageId).create();
+ if (initializeLastMessageIdInBroker) {
+ if (messageId == MessageId.earliest) {
+ assertTrue(reader.hasMessageAvailable());
+ } else {
+ assertFalse(reader.hasMessageAvailable());
+ }
+ } // else: lastMessageIdInBroker is earliest
+ reader.seek(timestampBeforeSend);
+ assertTrue(reader.hasMessageAvailable());
+ }
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 6c2ded819a5..5a0e5de330d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -218,6 +218,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
private final AtomicReference<ClientCnx> clientCnxUsedForConsumerRegistration = new AtomicReference<>();
private final List<Throwable> previousExceptions = new CopyOnWriteArrayList<Throwable>();
+ private volatile boolean hasSoughtByTimestamp = false;
static <T> ConsumerImpl<T> newConsumerImpl(PulsarClientImpl client,
String topic,
ConsumerConfigurationData<T> conf,
@@ -2252,7 +2253,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
new PulsarClientException("Only support seek by messageId or timestamp"));
}
- private CompletableFuture<Void> seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId, String seekBy) {
+ private CompletableFuture<Void> seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId,
+ Long seekTimestamp, String seekBy) {
AtomicLong opTimeoutMs = new AtomicLong(client.getConfiguration().getOperationTimeoutMs());
Backoff backoff = new BackoffBuilder()
.setInitialTime(100, TimeUnit.MILLISECONDS)
@@ -2269,11 +2271,11 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return FutureUtil.failedFuture(new IllegalStateException(message));
}
seekFuture = new CompletableFuture<>();
- seekAsyncInternal(requestId, seek, seekId, seekBy, backoff, opTimeoutMs);
+ seekAsyncInternal(requestId, seek, seekId, seekTimestamp, seekBy, backoff, opTimeoutMs);
return seekFuture;
}
- private void seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId, String seekBy,
+ private void seekAsyncInternal(long requestId, ByteBuf seek, MessageId seekId, Long seekTimestamp, String seekBy,
final Backoff backoff, final AtomicLong remainingTime) {
ClientCnx cnx = cnx();
if (isConnected() && cnx != null) {
@@ -2281,6 +2283,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
seekMessageId = (MessageIdAdv) seekId;
log.info("[{}][{}] Seeking subscription to {}", topic, subscription, seekBy);
+ final boolean originalHasSoughtByTimestamp = hasSoughtByTimestamp;
+ hasSoughtByTimestamp = (seekTimestamp != null);
cnx.sendRequestWithId(seek, requestId).thenRun(() -> {
log.info("[{}][{}] Successfully reset subscription to {}", topic, subscription, seekBy);
acknowledgmentsGroupingTracker.flushAndClean();
@@ -2304,6 +2308,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
}).exceptionally(e -> {
seekMessageId = originSeekMessageId;
+ hasSoughtByTimestamp = originalHasSoughtByTimestamp;
log.error("[{}][{}] Failed to reset subscription: {}", topic, subscription, e.getCause().getMessage());
failSeek(
@@ -2326,7 +2331,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
log.warn("[{}] [{}] Could not get connection while seek -- Will try again in {} ms",
topic, getHandlerName(), nextDelay);
remainingTime.addAndGet(-nextDelay);
- seekAsyncInternal(requestId, seek, seekId, seekBy, backoff, remainingTime);
+ seekAsyncInternal(requestId, seek, seekId, seekTimestamp, seekBy, backoff, remainingTime);
}, nextDelay, TimeUnit.MILLISECONDS);
}
}
@@ -2343,7 +2348,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
String seekBy = String.format("the timestamp %d", timestamp);
long requestId = client.newRequestId();
return seekAsyncInternal(requestId, Commands.newSeek(consumerId, requestId, timestamp),
- MessageId.earliest, seekBy);
+ MessageId.earliest, timestamp, seekBy);
}
@Override
@@ -2369,7 +2374,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
seek = Commands.newSeek(consumerId, requestId, msgId.getLedgerId(), msgId.getEntryId(), ackSetArr);
}
- return seekAsyncInternal(requestId, seek, messageId, seekBy);
+ return seekAsyncInternal(requestId, seek, messageId, null, seekBy);
}
public boolean hasMessageAvailable() throws PulsarClientException {
@@ -2389,13 +2394,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
// we haven't read yet. use startMessageId for comparison
if (lastDequeuedMessageId == MessageId.earliest) {
+ // If the last seek is called with timestamp, startMessageId cannot represent the position to start, so we
+ // have to get the mark-delete position from the GetLastMessageId response to compare as well.
// if we are starting from latest, we should seek to the actual last message first.
// allow the last one to be read when read head inclusively.
- if (MessageId.latest.equals(startMessageId)) {
-
+ final boolean hasSoughtByTimestamp = this.hasSoughtByTimestamp;
+ if (MessageId.latest.equals(startMessageId) || hasSoughtByTimestamp) {
CompletableFuture<GetLastMessageIdResponse> future = internalGetLastMessageIdAsync();
// if the consumer is configured to read inclusive then we need to seek to the last message
- if (resetIncludeHead) {
+ if (resetIncludeHead && !hasSoughtByTimestamp) {
future = future.thenCompose((lastMessageIdResponse) ->
seekAsync(lastMessageIdResponse.lastMessageId)
.thenApply((ignore) -> lastMessageIdResponse));