Skip to content

[fix][broker] Fix tableview divergence in ServiceUnitStateTableViewSyncer causing flaky tests #25946

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 3 commits into from
Jun 7, 2026
Merged

[fix][broker] Fix tableview divergence in ServiceUnitStateTableViewSyncer causing flaky tests #25946

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ffc679c
[fix][broker] Fix tableview divergence in ServiceUnitStateTableViewSy…
lhotari Jun 5, 2026
b862d2f
Replace "dummy" method with NOOP_CONSUMER
lhotari Jun 5, 2026
f713a5c
Refactor: remove duplication
lhotari Jun 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 149 additions & 37 deletions .../apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateTableViewSyncer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,20 +23,26 @@
import static org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.COMPACTION_THRESHOLD;
import static org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.configureSystemTopics;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.google.common.annotations.VisibleForTesting;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiConsumer;
import lombok.Cleanup;
import lombok.CustomLog;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.ObjectMapperFactory;
import org.jspecify.annotations.NonNull;

/**
* Defines ServiceUnitTableViewSyncer.
Expand All @@ -47,10 +53,15 @@
public class ServiceUnitStateTableViewSyncer implements Closeable {
private static final int MAX_CONCURRENT_SYNC_COUNT = 100;
private static final int SYNC_WAIT_TIME_IN_SECS = 300;
private static final long RECONCILE_INTERVAL_IN_MILLIS = 5_000;
private static final BiConsumer<String, ServiceUnitStateData> NOOP_CONSUMER = (__, ___) -> {
};
private volatile int syncWaitTimeInSecs = SYNC_WAIT_TIME_IN_SECS;
private PulsarService pulsar;
private volatile ServiceUnitStateTableView systemTopicTableView;
private volatile ServiceUnitStateTableView metadataStoreTableView;
private volatile boolean isActive = false;
private final ObjectWriter jsonWriter = ObjectMapperFactory.getMapper().writer();


public void start(PulsarService pulsar)
Expand Down Expand Up @@ -82,53 +93,77 @@ public void start(PulsarService pulsar)
}

private CompletableFuture<Void> syncToSystemTopic(String key, ServiceUnitStateData data) {
return systemTopicTableView.put(key, data);
return logIfFailed(sync(systemTopicTableView, key, data), key, data, "systemTopic");
}

private CompletableFuture<Void> syncToMetadataStore(String key, ServiceUnitStateData data) {
return metadataStoreTableView.put(key, data);
return logIfFailed(sync(metadataStoreTableView, key, data), key, data, "metadataStore");
}

private void dummy(String key, ServiceUnitStateData data) {
private CompletableFuture<Void> sync(ServiceUnitStateTableView dst, String key, ServiceUnitStateData data) {
// A null tail item is a tombstone: the source view removed the key. Route it to
// delete() rather than put(): the metadata-store view's put() rejects null
// (@NonNull) and the system-topic view's delete() is itself a null-valued put(),
// so a uniform delete keeps both sync directions symmetric and prevents a missed
// deletion from leaving the two views with different sizes (which would make
// waitUntilSynced spin until the timeout budget).
return data == null ? dst.delete(key) : dst.put(key, data);
}

private CompletableFuture<Void> logIfFailed(CompletableFuture<Void> future, String key,
ServiceUnitStateData data, String dst) {
return future.whenComplete((__, e) -> {
if (e != null && !(e instanceof PulsarClientException.AlreadyClosedException)) {
log.warn().attr("dst", dst).attr("serviceUnit", key).attr("data", data).exception(e)
.log("Failed to sync tableview item; sizes may diverge until the next update");
}
});
}

private void syncExistingItems()
throws IOException, ExecutionException, InterruptedException, TimeoutException {
long started = System.currentTimeMillis();

@Cleanup
ServiceUnitStateTableView metadataStoreTableView = new ServiceUnitStateMetadataStoreTableViewImpl();
metadataStoreTableView.start(
pulsar,
this::dummy,
this::dummy,
this::dummy
NOOP_CONSUMER,
NOOP_CONSUMER,
NOOP_CONSUMER
);

@Cleanup
ServiceUnitStateTableView systemTopicTableView = new ServiceUnitStateTableViewImpl();
systemTopicTableView.start(
pulsar,
this::dummy,
this::dummy,
this::dummy
NOOP_CONSUMER,
NOOP_CONSUMER,
NOOP_CONSUMER
);


var syncer = pulsar.getConfiguration().getLoadBalancerServiceUnitTableViewSyncer();
ServiceUnitStateTableView src;
ServiceUnitStateTableView dst;
if (syncer == SystemTopicToMetadataStoreSyncer) {
clean(metadataStoreTableView);
syncExistingItemsToMetadataStore(systemTopicTableView);
src = systemTopicTableView;
dst = metadataStoreTableView;
} else {
clean(systemTopicTableView);
syncExistingItemsToSystemTopic(metadataStoreTableView, systemTopicTableView);
src = metadataStoreTableView;
dst = systemTopicTableView;
}

if (!waitUntilSynced(metadataStoreTableView, systemTopicTableView, started)) {
if (!waitUntilSynced(src, dst, started)) {
throw new TimeoutException(
syncer + " failed to sync existing items in tableviews. MetadataStoreTableView.size: "
+ metadataStoreTableView.entrySet().size()
+ ", SystemTopicTableView.size: " + systemTopicTableView.entrySet().size() + " in "
+ SYNC_WAIT_TIME_IN_SECS + " secs");
+ syncWaitTimeInSecs + " secs");
}

log.info().attr("metadataStoreTableViewSize", metadataStoreTableView.entrySet().size())
Expand All @@ -154,27 +189,29 @@ private void syncTailItems() throws InterruptedException, IOException, TimeoutEx
this.metadataStoreTableView.start(
pulsar,
this::syncToSystemTopic,
this::dummy,
this::dummy
NOOP_CONSUMER,
NOOP_CONSUMER
);
log.info("Started MetadataStoreTableView");

this.systemTopicTableView = new ServiceUnitStateTableViewImpl();
this.systemTopicTableView.start(
pulsar,
this::syncToMetadataStore,
this::dummy,
this::dummy
NOOP_CONSUMER,
NOOP_CONSUMER
);
log.info("Started SystemTopicTableView");

var syncer = pulsar.getConfiguration().getLoadBalancerServiceUnitTableViewSyncer();
if (!waitUntilSynced(metadataStoreTableView, systemTopicTableView, started)) {
var src = syncer == SystemTopicToMetadataStoreSyncer ? systemTopicTableView : metadataStoreTableView;
var dst = syncer == SystemTopicToMetadataStoreSyncer ? metadataStoreTableView : systemTopicTableView;
if (!waitUntilSynced(src, dst, started)) {
throw new TimeoutException(
syncer + " failed to sync tableviews. MetadataStoreTableView.size: "
+ metadataStoreTableView.entrySet().size()
+ ", SystemTopicTableView.size: " + systemTopicTableView.entrySet().size() + " in "
+ SYNC_WAIT_TIME_IN_SECS + " secs");
+ syncWaitTimeInSecs + " secs");
}


Expand All @@ -187,62 +224,132 @@ private void syncTailItems() throws InterruptedException, IOException, TimeoutEx
private void syncExistingItemsToMetadataStore(ServiceUnitStateTableView src)
throws JsonProcessingException, ExecutionException, InterruptedException, TimeoutException {
// Directly use store to sync existing items to metadataStoreTableView(otherwise, they are conflicted out)
var store = pulsar.getLocalMetadataStore();
var writer = ObjectMapperFactory.getMapper().writer();
var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
List<CompletableFuture<Void>> futures = new ArrayList<>();
var srcIter = src.entrySet().iterator();
while (srcIter.hasNext()) {
var e = srcIter.next();
futures.add(store.put(ServiceUnitStateMetadataStoreTableViewImpl.PATH_PREFIX + "/" + e.getKey(),
writer.writeValueAsBytes(e.getValue()), Optional.empty()).thenApply(__ -> null));
if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !srcIter.hasNext()) {
FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
}
futures.add(writeToMetadataStore(e.getKey(), e.getValue()));
maybeWaitCompletion(futures, !srcIter.hasNext());
}
}

private void maybeWaitCompletion(List<CompletableFuture<Void>> futures, boolean forceWait)
throws InterruptedException, ExecutionException, TimeoutException {
if (!futures.isEmpty() && (futures.size() == MAX_CONCURRENT_SYNC_COUNT || forceWait)) {
FutureUtil.waitForAll(futures)
.get(pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds(), TimeUnit.SECONDS);
futures.clear();
}
}

private @NonNull CompletableFuture<Void> writeToMetadataStore(String key, ServiceUnitStateData value)
throws JsonProcessingException {
return pulsar.getLocalMetadataStore().put(ServiceUnitStateMetadataStoreTableViewImpl.PATH_PREFIX + "/" + key,
jsonWriter.writeValueAsBytes(value), Optional.empty()).thenApply(__ -> null);
}

private void syncExistingItemsToSystemTopic(ServiceUnitStateTableView src,
ServiceUnitStateTableView dst)
throws ExecutionException, InterruptedException, TimeoutException {
var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
List<CompletableFuture<Void>> futures = new ArrayList<>();
var srcIter = src.entrySet().iterator();
while (srcIter.hasNext()) {
var e = srcIter.next();
futures.add(dst.put(e.getKey(), e.getValue()));
if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !srcIter.hasNext()) {
FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
}
maybeWaitCompletion(futures, !srcIter.hasNext());
}
}

private void clean(ServiceUnitStateTableView dst)
throws ExecutionException, InterruptedException, TimeoutException {
var opTimeout = pulsar.getConfiguration().getMetadataStoreOperationTimeoutSeconds();
var dstIter = dst.entrySet().iterator();
List<CompletableFuture<Void>> futures = new ArrayList<>();
while (dstIter.hasNext()) {
var e = dstIter.next();
futures.add(dst.delete(e.getKey()));
if (futures.size() == MAX_CONCURRENT_SYNC_COUNT || !dstIter.hasNext()) {
FutureUtil.waitForAll(futures).get(opTimeout, TimeUnit.SECONDS);
}
maybeWaitCompletion(futures, !dstIter.hasNext());
}
}

private boolean waitUntilSynced(ServiceUnitStateTableView srt, ServiceUnitStateTableView dst, long started)
private boolean waitUntilSynced(ServiceUnitStateTableView src, ServiceUnitStateTableView dst, long started)
throws InterruptedException {
while (srt.entrySet().size() != dst.entrySet().size()) {
if (TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - started)
> SYNC_WAIT_TIME_IN_SECS) {
long lastReconciled = started;
while (src.entrySet().size() != dst.entrySet().size()) {
long now = System.currentTimeMillis();
if (TimeUnit.MILLISECONDS.toSeconds(now - started) > syncWaitTimeInSecs) {
return false;
}
// Give in-flight syncs a grace period to settle on their own, then reconcile
// periodically: updates that raced with the table views' (re)start were replayed
// to the fresh views as existing items — which are deliberately not wired to
// sync — so without reconciliation the views would never converge.
if (now - lastReconciled >= RECONCILE_INTERVAL_IN_MILLIS) {
log.debug().attr("srcSize", src.entrySet().size()).attr("dstSize", dst.entrySet().size())
.attr("elapsedSecs", TimeUnit.MILLISECONDS.toSeconds(now - started))
.log("Tableviews not synced yet; reconciling");
reconcile(src, dst, started);
lastReconciled = now;
}
Thread.sleep(100);
}
return true;
}

/**
* Copies items the destination table view is missing and removes stale items that no longer
* exist in the source. Channel updates that land between the existing-items copy and the
* registration of the tail listeners are only visible as existing items of the freshly
* started views, so the tail listeners never see them. Writes flow to the migration source
* while the syncer starts, making the source view authoritative; destination-only items are
* removed only when they predate this sync phase and are still absent from the source, so a
* concurrent fresh write to the destination is never discarded. Failures are logged and left
* for the next reconcile pass. Runs on the caller's (load manager) thread with each batch
* bounded by the metadata store operation timeout.
*/
private void reconcile(ServiceUnitStateTableView src, ServiceUnitStateTableView dst, long started)
throws InterruptedException {
// Snapshot the destination entries before iterating the source so that a key arriving
// in the destination through a concurrent tail sync cannot be misclassified as stale.
var staleDstItems = new HashMap<String, ServiceUnitStateData>();
for (var e : dst.entrySet()) {
staleDstItems.put(e.getKey(), e.getValue());
}
try {
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (var e : src.entrySet()) {
if (staleDstItems.remove(e.getKey()) == null) {
log.info().attr("serviceUnit", e.getKey())
.log("Reconciling item missing from the destination tableview");
if (dst.isMetadataStoreBased()) {
// Write directly to the store like syncExistingItemsToMetadataStore
// does; the view's put() would conflict the item out.
futures.add(writeToMetadataStore(e.getKey(), e.getValue()));
} else {
futures.add(dst.put(e.getKey(), e.getValue()));
}
maybeWaitCompletion(futures, false);
}
}
for (var e : staleDstItems.entrySet()) {
// Only remove items written before this sync phase began and re-confirmed absent
// from the source: a fresh destination write (e.g. from a broker already switched
// to the destination implementation) is propagated to the source by the tail
// listener instead of being deleted here.
if (e.getValue().timestamp() < started && src.get(e.getKey()) == null) {
Comment thread
lhotari marked this conversation as resolved.
log.info().attr("serviceUnit", e.getKey())
.log("Reconciling stale item in the destination tableview");
futures.add(dst.delete(e.getKey()));
maybeWaitCompletion(futures, false);
}
}
maybeWaitCompletion(futures, true);
} catch (IOException | ExecutionException | TimeoutException e) {
// Transient write failures leave a size divergence behind; the next reconcile pass
// (or the sync-wait timeout) handles it.
log.warn().exception(e).log("Failed to reconcile tableview items");
}
}

@Override
public void close() throws IOException {
if (!isActive) {
Expand Down Expand Up @@ -282,4 +389,9 @@ public void close() throws IOException {
public boolean isActive() {
return isActive;
}

@VisibleForTesting
public void setSyncWaitTimeInSecs(int syncWaitTimeInSecs) {
this.syncWaitTimeInSecs = syncWaitTimeInSecs;
}
}
16 changes: 15 additions & 1 deletion ...va/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImplBaseTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.awaitility.Awaitility;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
Expand Down Expand Up @@ -197,7 +198,20 @@ protected void cleanup() throws Exception {

@BeforeMethod(alwaysRun = true)
protected void initializeState() throws PulsarAdminException, IllegalAccessException {
admin.namespaces().unload(defaultTestNamespace);
// After a prior test churned leader election, the channel-topic bundle can be left
// unserved ("not served by this instance"), making the unload's channel publish fail
// (HTTP 500) or hang server-side until the background monitor task (120s interval)
// reconciles the brokers' roles with the channel ownership. Drive monitor() eagerly to
// heal that state, bound each unload attempt (a synchronous unload() can block longer
// than the whole retry window), and fail loudly on exhaustion.
Awaitility.await().atMost(120, TimeUnit.SECONDS)
.pollInterval(1, TimeUnit.SECONDS)
.ignoreExceptions()
.untilAsserted(() -> {
primaryLoadManager.monitor();
secondaryLoadManager.monitor();
admin.namespaces().unloadAsync(defaultTestNamespace).get(15, TimeUnit.SECONDS);
});
reset(primaryLoadManager, secondaryLoadManager);
FieldUtils.writeDeclaredField(pulsarClient, "lookup", lookupService, true);
pulsar1.getConfig().setLoadBalancerMultiPhaseBundleUnload(true);
Expand Down
Loading
Loading