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 2026/10/21 19:45:39 UTC
(pulsar) 09/21: [fix][client] Complete a TableView refresh waiting for a message that could not be applied (#26986)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit b7c5d33f3645afb2f70d3a355f050625e9646684
Author: SeungTaek Lim <na...@gmail.com>
AuthorDate: Wed Oct 21 08:57:57 2026 +0900
[fix][client] Complete a TableView refresh waiting for a message that could not be applied (#26986)
(cherry picked from commit 08864bc55a4eab1e4218235784a483a4e063c62b)
---
.../pulsar/client/impl/AbstractTableViewImpl.java | 16 +++++--
.../pulsar/client/impl/TableViewImplTest.java | 56 ++++++++++++++++++++++
2 files changed, 68 insertions(+), 4 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java
index dc6c46750adb..12ca5507cac1 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/AbstractTableViewImpl.java
@@ -280,11 +280,19 @@ abstract class AbstractTableViewImpl<T, V> implements TableView<V> {
private void handleMessage(Message<T> msg) {
try {
- if (msg.hasKey()) {
- handleKeyedMessage(msg);
- } else {
- // Keyless messages also advance the refresh position.
+ try {
+ if (msg.hasKey()) {
+ handleKeyedMessage(msg);
+ } else {
+ // Keyless messages also advance the refresh position.
+ lastReadPositions.put(msg.getTopicName(), msg.getMessageId());
+ }
+ } catch (Throwable t) {
+ // The message is skipped, so it is processed as far as a refresh is concerned: one waiting for it
+ // completes now instead of waiting for the next message, which may never come on a quiet topic.
lastReadPositions.put(msg.getTopicName(), msg.getMessageId());
+ checkAllFreshTask(msg);
+ throw t;
}
checkAllFreshTask(msg);
} finally {
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewImplTest.java
index 5b8dffcd703b..e1652e513a43 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/TableViewImplTest.java
@@ -803,6 +803,62 @@ public class TableViewImplTest {
}
}
+ @Test(timeOut = 10_000)
+ @SuppressWarnings("unchecked")
+ public void testRefreshWaitingForAMessageThatCouldNotBeAppliedCompletes() throws Exception {
+ TopicCompactionStrategy<String> strategy = mock(TopicCompactionStrategy.class);
+ when(strategy.shouldKeepLeft(any(), any())).thenThrow(new IllegalStateException("strategy failure"));
+ try (TailRetryFixture f = new TailRetryFixture(strategy)) {
+ CompletableFuture<Message<String>> read = new CompletableFuture<>();
+ when(f.reader.readNextAsync()).thenReturn(read, new CompletableFuture<>());
+ f.tableView.start().get(5, TimeUnit.SECONDS);
+ CompletableFuture<Void> refresh = f.pendingRefresh();
+ Message<String> message = mock(Message.class);
+ when(message.getTopicName()).thenReturn(TAIL_RETRY_TOPIC);
+ when(message.getMessageId()).thenReturn(new MessageIdImpl(1, 5, -1));
+ when(message.hasKey()).thenReturn(true);
+ when(message.getKey()).thenReturn("key");
+ when(message.size()).thenReturn(1);
+ when(message.getValue()).thenReturn("value");
+
+ // The message the refresh is waiting for is skipped.
+ read.complete(message);
+
+ assertTrue(refresh.isDone(), "A skipped message is processed: the refresh must not wait for the next one");
+ refresh.get(5, TimeUnit.SECONDS);
+ assertNull(f.tableView.get("key"), "The skipped message must not update the table");
+ assertEquals(f.retries.size(), 1, "The tail reads must go on after the skipped message");
+ }
+ }
+
+ @Test(timeOut = 10_000)
+ @SuppressWarnings("unchecked")
+ public void testRefreshStartedAfterASkippedMessageDoesNotWaitForIt() throws Exception {
+ TopicCompactionStrategy<String> strategy = mock(TopicCompactionStrategy.class);
+ when(strategy.shouldKeepLeft(any(), any())).thenThrow(new IllegalStateException("strategy failure"));
+ try (TailRetryFixture f = new TailRetryFixture(strategy)) {
+ TopicMessageId messageId = new TopicMessageIdImpl(TAIL_RETRY_TOPIC, new MessageIdImpl(1, 5, -1));
+ Message<String> message = mock(Message.class);
+ when(message.getTopicName()).thenReturn(TAIL_RETRY_TOPIC);
+ when(message.getMessageId()).thenReturn(messageId);
+ when(message.hasKey()).thenReturn(true);
+ when(message.getKey()).thenReturn("key");
+ when(message.size()).thenReturn(1);
+ when(message.getValue()).thenReturn("value");
+ when(f.reader.readNextAsync()).thenReturn(CompletableFuture.completedFuture(message),
+ new CompletableFuture<>());
+ // The message is skipped while the table view starts.
+ f.tableView.start().get(5, TimeUnit.SECONDS);
+
+ // A refresh started afterwards looks up that message as the last one on the topic.
+ when(f.reader.getLastMessageIdsAsync()).thenReturn(CompletableFuture.completedFuture(List.of(messageId)));
+ CompletableFuture<Void> refresh = f.tableView.refreshAsync();
+
+ assertTrue(refresh.isDone(), "The skipped message is processed: a refresh must not wait for it");
+ refresh.get(5, TimeUnit.SECONDS);
+ }
+ }
+
/**
* Captures the WARN and above events of the table view's logger that are logged on the calling thread. The
* tail reads of a {@link TailRetryFixture} run on the test thread, while a retry left over by another test