You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/10/10 07:17:43 UTC
(pulsar) branch master updated: [fix][client] Apply topic changes that arrive while a V5 namespace consumer is created (#26840)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 2992c62fbc7b [fix][client] Apply topic changes that arrive while a V5 namespace consumer is created (#26840)
2992c62fbc7b is described below
commit 2992c62fbc7be971a762e9531e51007772238b48
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Sat Oct 10 00:17:35 2026 -0700
[fix][client] Apply topic changes that arrive while a V5 namespace consumer is created (#26840)
---
...mespaceConsumerTopicChangeDuringCreateTest.java | 264 +++++++++++++++++++++
.../client/impl/v5/ScalableTopicsWatcher.java | 105 ++++++--
.../client/impl/v5/ScalableTopicsWatcherTest.java | 186 +++++++++++++++
3 files changed, 538 insertions(+), 17 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicChangeDuringCreateTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicChangeDuringCreateTest.java
new file mode 100644
index 000000000000..147b62651e7f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/v5/V5NamespaceConsumerTopicChangeDuringCreateTest.java
@@ -0,0 +1,264 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.time.Duration;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import lombok.Cleanup;
+import lombok.CustomLog;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.v5.Message;
+import org.apache.pulsar.client.api.v5.Producer;
+import org.apache.pulsar.client.api.v5.QueueConsumer;
+import org.apache.pulsar.client.api.v5.StreamConsumer;
+import org.apache.pulsar.client.api.v5.V5ClientBaseTest;
+import org.apache.pulsar.client.api.v5.config.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.v5.schema.Schema;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.ConnectionPool;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.client.impl.ScalableTopicsWatcherSession;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
+import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.scalable.SegmentTopicName;
+import org.apache.pulsar.common.util.netty.EventLoopUtil;
+import org.awaitility.Awaitility;
+import org.testng.annotations.Test;
+
+/**
+ * A scalable topic created or deleted in a namespace while a namespace consumer is being created must be
+ * attached or detached once the consumer exists. The broker reports each change once, and the consumer
+ * follows the changes only after it has subscribed to every topic of the initial snapshot.
+ *
+ * <p>The tests hold back the segment subscriptions of a topic in the initial snapshot, change the
+ * namespace, and wait until the consumer's watcher has received the change before letting the creation
+ * finish.
+ */
+public class V5NamespaceConsumerTopicChangeDuringCreateTest extends V5ClientBaseTest {
+
+ /**
+ * A v4 client that holds back the segment subscriptions to the topics the test picks until it allows
+ * them, records the scalable topics it subscribes to, and keeps the namespace watchers registered on
+ * its connections.
+ */
+ @CustomLog
+ private static final class HoldingClient extends PulsarClientImpl {
+
+ private final List<ScalableTopicsWatcher> watchers;
+ private final Set<String> heldTopics = ConcurrentHashMap.newKeySet();
+ private final CompletableFuture<Void> attachAllowed = new CompletableFuture<>();
+ private final List<String> attachRequests = new CopyOnWriteArrayList<>();
+ private final Set<String> attachedTopics = ConcurrentHashMap.newKeySet();
+
+ static HoldingClient create(String serviceUrl) throws Exception {
+ ClientConfigurationData conf = new ClientConfigurationData();
+ conf.setServiceUrl(serviceUrl);
+ conf.setStatsIntervalSeconds(0);
+ EventLoopGroup eventLoopGroup =
+ EventLoopUtil.newEventLoopGroup(1, false, new DefaultThreadFactory("holding-client-io"));
+ List<ScalableTopicsWatcher> watchers = new CopyOnWriteArrayList<>();
+ ConnectionPool connectionPool = new ConnectionPool(InstrumentProvider.NOOP, conf, eventLoopGroup,
+ () -> new ClientCnx(InstrumentProvider.NOOP, conf, eventLoopGroup) {
+ @Override
+ public void registerScalableTopicsWatcher(long watchId,
+ ScalableTopicsWatcherSession watcher) {
+ watchers.add((ScalableTopicsWatcher) watcher);
+ super.registerScalableTopicsWatcher(watchId, watcher);
+ }
+ }, null);
+ return new HoldingClient(conf, eventLoopGroup, connectionPool, watchers);
+ }
+
+ private HoldingClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup,
+ ConnectionPool connectionPool, List<ScalableTopicsWatcher> watchers)
+ throws Exception {
+ super(conf, eventLoopGroup, connectionPool);
+ this.watchers = watchers;
+ }
+
+ void hold(String topic) {
+ heldTopics.add(topic);
+ }
+
+ void allowAttach() {
+ attachAllowed.complete(null);
+ }
+
+ /** The scalable topics of the segments subscribed to so far, whether held or not. */
+ List<String> attachRequests() {
+ return List.copyOf(attachRequests);
+ }
+
+ /** The scalable topics with a segment subscription completed. */
+ Set<String> attachedTopics() {
+ return Set.copyOf(attachedTopics);
+ }
+
+ /** The watcher of the namespace consumer: the only one this client opens. */
+ ScalableTopicsWatcher watcher() {
+ Awaitility.await().until(() -> !watchers.isEmpty());
+ return watchers.get(0);
+ }
+
+ // The v4 Schema is qualified: the V5 one is imported.
+ @Override
+ public <T> CompletableFuture<Consumer<T>> subscribeSegmentAsync(
+ ConsumerConfigurationData<T> conf, org.apache.pulsar.client.api.Schema<T> schema) {
+ String topic = SegmentTopicName.getParentTopicName(TopicName.get(conf.getSingleTopic())).toString();
+ attachRequests.add(topic);
+ CompletableFuture<Void> allowed =
+ heldTopics.contains(topic) ? attachAllowed : CompletableFuture.completedFuture(null);
+ return allowed.thenCompose(__ -> super.subscribeSegmentAsync(conf, schema))
+ .thenApply(consumer -> {
+ attachedTopics.add(topic);
+ return consumer;
+ });
+ }
+
+ // The connection pool and its I/O threads were passed in, so the client doesn't close them.
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return super.closeAsync().whenComplete((__, ___) -> {
+ try {
+ getCnxPool().close();
+ } catch (Exception e) {
+ log.warn().exception(e).log("Failed to close the connection pool");
+ }
+ eventLoopGroup.shutdownGracefully();
+ });
+ }
+ }
+
+ @Test
+ public void testQueueConsumerAttachesTopicCreatedDuringCreate() throws Exception {
+ String existing = newScalableTopic(1);
+ PulsarClientV5 client = newHoldingClient();
+ HoldingClient holding = (HoldingClient) client.v4Client();
+ holding.hold(existing);
+ CompletableFuture<QueueConsumer<String>> creating = client.newQueueConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribeAsync();
+ Awaitility.await().until(() -> holding.attachRequests().contains(existing));
+ ScalableTopicsWatcher watcher = holding.watcher();
+
+ String created = newScalableTopic(1);
+ Awaitility.await().until(() -> watcher.currentSetForTesting().contains(created));
+ assertThat(creating).as("the consumer is still subscribing to the existing topic").isNotDone();
+ holding.allowAttach();
+ @Cleanup
+ QueueConsumer<String> consumer = creating.get(30, SECONDS);
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(((MultiTopicQueueConsumer<String>) consumer).attachedTopicsForTesting())
+ .as("topics the consumer is attached to")
+ .containsExactlyInAnyOrder(existing, created));
+ send(created, "created-during-create");
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message sent to the topic created during the creation").isNotNull();
+ assertThat(msg.value()).isEqualTo("created-during-create");
+ }
+
+ @Test
+ public void testQueueConsumerDetachesTopicDeletedDuringCreate() throws Exception {
+ String kept = newScalableTopic(1);
+ String deleted = newScalableTopic(1);
+ PulsarClientV5 client = newHoldingClient();
+ HoldingClient holding = (HoldingClient) client.v4Client();
+ holding.hold(kept);
+ CompletableFuture<QueueConsumer<String>> creating = client.newQueueConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribeAsync();
+ Awaitility.await().until(() -> holding.attachRequests().contains(kept)
+ && holding.attachedTopics().contains(deleted));
+ ScalableTopicsWatcher watcher = holding.watcher();
+
+ admin.scalableTopics().deleteScalableTopic(deleted, true);
+ Awaitility.await().until(() -> !watcher.currentSetForTesting().contains(deleted));
+ assertThat(creating).as("the consumer is still subscribing to the kept topic").isNotDone();
+ holding.allowAttach();
+ @Cleanup
+ QueueConsumer<String> consumer = creating.get(30, SECONDS);
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(((MultiTopicQueueConsumer<String>) consumer).attachedTopicsForTesting())
+ .as("topics the consumer is attached to")
+ .containsExactly(kept));
+ }
+
+ @Test
+ public void testStreamConsumerReceivesFromTopicCreatedDuringCreate() throws Exception {
+ String existing = newScalableTopic(1);
+ PulsarClientV5 client = newHoldingClient();
+ HoldingClient holding = (HoldingClient) client.v4Client();
+ holding.hold(existing);
+ CompletableFuture<StreamConsumer<String>> creating = client.newStreamConsumer(Schema.string())
+ .namespace(getNamespace())
+ .subscriptionName("sub")
+ .subscriptionInitialPosition(SubscriptionInitialPosition.EARLIEST)
+ .subscribeAsync();
+ Awaitility.await().until(() -> holding.attachRequests().contains(existing));
+ ScalableTopicsWatcher watcher = holding.watcher();
+
+ String created = newScalableTopic(1);
+ Awaitility.await().until(() -> watcher.currentSetForTesting().contains(created));
+ assertThat(creating).as("the consumer is still subscribing to the existing topic").isNotDone();
+ holding.allowAttach();
+ @Cleanup
+ StreamConsumer<String> consumer = creating.get(30, SECONDS);
+
+ Awaitility.await().untilAsserted(() ->
+ assertThat(holding.attachedTopics())
+ .as("topics the consumer is attached to")
+ .containsExactlyInAnyOrder(existing, created));
+ send(created, "created-during-create");
+ Message<String> msg = consumer.receive(Duration.ofSeconds(10));
+ assertThat(msg).as("message sent to the topic created during the creation").isNotNull();
+ assertThat(msg.value()).isEqualTo("created-during-create");
+ }
+
+ // --- Helpers ---
+
+ private PulsarClientV5 newHoldingClient() throws Exception {
+ return track(new PulsarClientV5(HoldingClient.create(getBrokerServiceUrl()), "holding-client", null));
+ }
+
+ private void send(String topic, String value) throws Exception {
+ @Cleanup
+ Producer<String> producer = v5Client.newProducer(Schema.string())
+ .topic(topic)
+ .create();
+ producer.newMessage()
+ .value(value)
+ .send();
+ }
+}
diff --git a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
index 3ece8fe0628c..7ac0c0160667 100644
--- a/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
+++ b/pulsar-client-v5/src/main/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcher.java
@@ -20,13 +20,16 @@ package org.apache.pulsar.client.impl.v5;
import io.github.merlimat.slog.Logger;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.ClientCnx;
@@ -58,7 +61,8 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
/**
* Listener for membership events. The watcher delivers events on the netty IO
- * thread; implementations should not block.
+ * thread, or on the thread setting the listener for the events received before it;
+ * never two at once. Implementations should not block.
*/
interface Listener {
/** Full set; replace any local state. */
@@ -84,6 +88,21 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
* hash may be read on a reconnect callback running elsewhere.
*/
private final Set<String> currentSet = Collections.synchronizedSet(new HashSet<>());
+ /**
+ * The changes to {@link #currentSet} the listener hasn't been handed yet: at first those
+ * since the initial snapshot, which the caller of {@link #start} handles itself before
+ * setting the listener. A topic added and removed again in between is in neither set; one
+ * removed and added again is in both. Guarded by {@code currentSet}.
+ */
+ private final Set<String> pendingAdded = new LinkedHashSet<>();
+ private final Set<String> pendingRemoved = new LinkedHashSet<>();
+ /**
+ * Whether a later snapshot replaced {@link #currentSet} since the listener was last
+ * notified, so that it gets the whole set. Guarded by {@code currentSet}.
+ */
+ private boolean pendingSnapshot;
+ /** Calls to {@link #notifyListener} not served yet; see there. */
+ private final AtomicInteger pendingNotifications = new AtomicInteger();
private volatile Listener listener;
private volatile ClientCnx cnx;
private volatile boolean closed = false;
@@ -163,25 +182,23 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
// Reset backoff on every successful snapshot — that's the broker confirming
// the session is live and our local state is consistent.
reconnectBackoff.reset();
- // Replace local set so the next reconnect computes the right hash.
+ boolean initial = !initialSnapshotFuture.isDone();
+ // Replace local set so the next reconnect computes the right hash. The snapshot
+ // supersedes any change the listener hasn't been handed yet.
synchronized (currentSet) {
currentSet.clear();
currentSet.addAll(topics);
+ pendingAdded.clear();
+ pendingRemoved.clear();
+ pendingSnapshot = !initial;
}
- if (!initialSnapshotFuture.isDone()) {
+ if (initial) {
initialSnapshotFuture.complete(topics);
// The listener is set by the caller AFTER start() resolves, so the initial
// snapshot is delivered via the future, not via onSnapshot's fan-out.
return;
}
- Listener l = listener;
- if (l != null) {
- try {
- l.onSnapshot(topics);
- } catch (Exception e) {
- log.error().exception(e).log("Listener threw on snapshot");
- }
- }
+ notifyListener();
}
@Override
@@ -191,13 +208,65 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
}
log.info().attr("added", added.size()).attr("removed", removed.size())
.log("Diff received");
- // Apply removed before added — covers rapid remove-then-add of the same name.
+ // Apply removed before added — covers rapid remove-then-add of the same name. Only
+ // what changes the set is a change for the listener: the broker can report a topic of
+ // the initial snapshot as added again.
+ synchronized (currentSet) {
+ for (String topic : removed) {
+ // A topic added since the listener was last notified: it never hears of it.
+ if (currentSet.remove(topic) && !pendingSnapshot && !pendingAdded.remove(topic)) {
+ pendingRemoved.add(topic);
+ }
+ }
+ for (String topic : added) {
+ if (currentSet.add(topic) && !pendingSnapshot) {
+ pendingAdded.add(topic);
+ }
+ }
+ }
+ notifyListener();
+ }
+
+ /**
+ * Hand the listener the changes it hasn't had yet. Both the I/O thread, on every event,
+ * and {@link #setListener} call this, so the listener calls are serialized here without
+ * holding a lock while the listener runs: the call that finds no other in progress
+ * delivers, and delivers again for each call made meanwhile. The changes that arrive while
+ * there is no listener, or while it runs, reach it together in its next call.
+ */
+ private void notifyListener() {
+ if (pendingNotifications.getAndIncrement() != 0) {
+ return;
+ }
+ do {
+ Listener l = listener;
+ if (l != null) {
+ deliverPending(l);
+ }
+ } while (pendingNotifications.decrementAndGet() != 0);
+ }
+
+ private void deliverPending(Listener l) {
+ List<String> snapshot = null;
+ List<String> added;
+ List<String> removed;
synchronized (currentSet) {
- currentSet.removeAll(removed);
- currentSet.addAll(added);
+ if (pendingSnapshot) {
+ pendingSnapshot = false;
+ snapshot = new ArrayList<>(currentSet);
+ }
+ added = List.copyOf(pendingAdded);
+ removed = List.copyOf(pendingRemoved);
+ pendingAdded.clear();
+ pendingRemoved.clear();
}
- Listener l = listener;
- if (l != null) {
+ if (snapshot != null) {
+ try {
+ l.onSnapshot(snapshot);
+ } catch (Exception e) {
+ log.error().exception(e).log("Listener threw on snapshot");
+ }
+ } else if (!added.isEmpty() || !removed.isEmpty()) {
try {
l.onDiff(added, removed);
} catch (Exception e) {
@@ -312,10 +381,12 @@ final class ScalableTopicsWatcher implements ScalableTopicsWatcherSession, AutoC
/**
* Set the listener that receives {@code Snapshot} / {@code Diff} events. Should
* be called after {@link #start()} resolves — the initial snapshot is delivered
- * via that future, not via the listener.
+ * via that future, not via the listener. The changes received since then are
+ * handed to the listener right away: the broker reports each change only once.
*/
void setListener(Listener listener) {
this.listener = listener;
+ notifyListener();
}
/**
diff --git a/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcherTest.java b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcherTest.java
new file mode 100644
index 000000000000..874176b3efc0
--- /dev/null
+++ b/pulsar-client-v5/src/test/java/org/apache/pulsar/client/impl/v5/ScalableTopicsWatcherTest.java
@@ -0,0 +1,186 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl.v5;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.client.impl.PulsarClientImpl;
+import org.apache.pulsar.common.naming.NamespaceName;
+import org.testng.annotations.Test;
+
+public class ScalableTopicsWatcherTest {
+
+ private static final String A = "topic://public/default/a";
+ private static final String B = "topic://public/default/b";
+ private static final String C = "topic://public/default/c";
+
+ /** Records the listener calls, one line each. */
+ private static class RecordingListener implements ScalableTopicsWatcher.Listener {
+ final List<String> calls = new CopyOnWriteArrayList<>();
+
+ @Override
+ public void onSnapshot(List<String> topics) {
+ calls.add("snapshot " + topics.stream().sorted().toList());
+ }
+
+ @Override
+ public void onDiff(List<String> added, List<String> removed) {
+ calls.add("added " + added + " removed " + removed);
+ }
+ }
+
+ /** A started watcher that has received its initial snapshot, of topics A and B. */
+ private static ScalableTopicsWatcher watcherWithSnapshot() {
+ PulsarClientImpl v4Client = mock(PulsarClientImpl.class);
+ // The broker's events are fed in by the tests: the connection never opens.
+ when(v4Client.getAnyBrokerProxyConnection()).thenReturn(new CompletableFuture<>());
+ ScalableTopicsWatcher watcher =
+ new ScalableTopicsWatcher(v4Client, NamespaceName.get("public/default"), Map.of());
+ CompletableFuture<List<String>> initial = watcher.start();
+ watcher.onSnapshot(List.of(A, B));
+ assertThat(initial).isCompletedWithValue(List.of(A, B));
+ return watcher;
+ }
+
+ @Test
+ public void testSetListenerDeliversChangesReceivedBeforeIt() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ // C is created and B deleted while the caller attaches to A and B.
+ watcher.onDiff(List.of(C), List.of());
+ watcher.onDiff(List.of(), List.of(B));
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+
+ assertThat(listener.calls).containsExactly("added [" + C + "] removed [" + B + "]");
+ }
+
+ @Test
+ public void testSetListenerDeliversNothingWhenNothingChangedSinceStart() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+ assertThat(listener.calls).isEmpty();
+
+ watcher.onDiff(List.of(C), List.of());
+ assertThat(listener.calls).containsExactly("added [" + C + "] removed []");
+ }
+
+ @Test
+ public void testTopicAddedAndRemovedBeforeSetListenerIsNotDelivered() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ watcher.onDiff(List.of(C), List.of());
+ watcher.onDiff(List.of(), List.of(C));
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+
+ assertThat(listener.calls).isEmpty();
+ }
+
+ @Test
+ public void testTopicRemovedAndAddedBeforeSetListenerIsDeliveredAsBoth() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ // B is deleted and created again: the listener must drop what it attached to and attach again.
+ watcher.onDiff(List.of(), List.of(B));
+ watcher.onDiff(List.of(B), List.of());
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+
+ assertThat(listener.calls).containsExactly("added [" + B + "] removed [" + B + "]");
+ }
+
+ @Test
+ public void testTopicOfSnapshotAddedAgainThenRemovedBeforeSetListenerIsDeliveredAsRemoved() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ // The broker can report a topic of the initial snapshot as added again; B is then deleted.
+ watcher.onDiff(List.of(B), List.of());
+ watcher.onDiff(List.of(), List.of(B));
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+
+ assertThat(listener.calls).containsExactly("added [] removed [" + B + "]");
+ }
+
+ @Test
+ public void testSnapshotBeforeSetListenerIsDeliveredWhole() {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ watcher.onDiff(List.of(C), List.of());
+ // A reconnect resyncs the set, and a change follows.
+ watcher.onSnapshot(List.of(A, C));
+ watcher.onDiff(List.of(B), List.of(A));
+ RecordingListener listener = new RecordingListener();
+
+ watcher.setListener(listener);
+
+ assertThat(listener.calls).containsExactly("snapshot [" + B + ", " + C + "]");
+ assertThat(watcher.currentSetForTesting()).containsExactlyInAnyOrder(B, C);
+ }
+
+ @Test(timeOut = 30_000)
+ public void testChangeArrivingWhileListenerRunsIsDeliveredAfterIt() throws Exception {
+ ScalableTopicsWatcher watcher = watcherWithSnapshot();
+ watcher.onDiff(List.of(C), List.of());
+ CountDownLatch delivering = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ RecordingListener listener = new RecordingListener() {
+ @Override
+ public void onDiff(List<String> added, List<String> removed) {
+ super.onDiff(added, removed);
+ delivering.countDown();
+ awaitUninterruptibly(resume);
+ }
+ };
+ Thread caller = new Thread(() -> watcher.setListener(listener));
+ caller.start();
+ try {
+ assertThat(delivering.await(10, TimeUnit.SECONDS)).isTrue();
+
+ // B is deleted, on the I/O thread, while the caller's thread is still delivering C: the I/O
+ // thread must neither wait for that delivery nor call the listener alongside it.
+ watcher.onDiff(List.of(), List.of(B));
+ assertThat(listener.calls).containsExactly("added [" + C + "] removed []");
+ } finally {
+ // Let the caller's thread finish even if an assertion failed.
+ resume.countDown();
+ caller.join();
+ }
+ assertThat(listener.calls).containsExactly(
+ "added [" + C + "] removed []",
+ "added [] removed [" + B + "]");
+ }
+
+ private static void awaitUninterruptibly(CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}