Skip to content
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

[fix][broker] Immediately tombstone Deleted and Free state bundles #22743

Merged
merged 4 commits into from
May 21, 2024
Merged

[fix][broker] Immediately tombstone Deleted and Free state bundles #22743

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d7ffdfc
[fix][broker] Tombsotne Deleted and Free state bundles
heesung-sn May 18, 2024
83fe613
updated split manager
heesung-sn May 18, 2024
03f513b
reverted multi phase unloading
heesung-sn May 19, 2024
4dcd4ac
updated the tombstone logic
heesung-sn May 20, 2024
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
Prev Previous commit
Next Next commit
Loading
Loading status checks…
updated split manager
@heesung-sn
heesung-sn committed May 18, 2024

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature. The key has expired.
GPG key ID: 4AEE18F83AFDEB23
Expired
Verified
Learn about vigilant mode
commit 83fe613770048301a26eae86717d3aaaa780df65
2 changes: 1 addition & 1 deletion ...r/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/manager/SplitManager.java
View file
Original file line number Diff line number Diff line change
@@ -97,7 +97,7 @@ public void handleEvent(String serviceUnit, ServiceUnitStateData data, Throwable
return;
}
switch (state) {
case Deleted, Owned, Init -> this.complete(serviceUnit, t);
case Init -> this.complete(serviceUnit, t);
default -> {
if (log.isDebugEnabled()) {
log.debug("Handling {} for service unit {}", data, serviceUnit);
27 changes: 5 additions & 22 deletions ...c/test/java/org/apache/pulsar/broker/loadbalance/extensions/manager/SplitManagerTest.java
View file
Original file line number Diff line number Diff line change
@@ -123,40 +123,23 @@ public void testSuccess() throws IllegalAccessException, ExecutionException, Int
manager.handleEvent(bundle,
new ServiceUnitStateData(ServiceUnitState.Free, dstBroker, VERSION_ID_INIT), null);
assertEquals(inFlightUnloadRequests.size(), 1);
assertEquals(counter.toMetrics(null).toString(),
counterExpected.toMetrics(null).toString());

manager.handleEvent(bundle,
new ServiceUnitStateData(ServiceUnitState.Deleted, dstBroker, VERSION_ID_INIT), null);
counterExpected.update(SplitDecision.Label.Success, Sessions);
assertEquals(inFlightUnloadRequests.size(), 0);
assertEquals(counter.toMetrics(null).toString(),
counterExpected.toMetrics(null).toString());
assertEquals(inFlightUnloadRequests.size(), 1);

// Success with Init state.
future = manager.waitAsync(CompletableFuture.completedFuture(null),
bundle, decision, 5, TimeUnit.SECONDS);
inFlightUnloadRequests = getinFlightUnloadRequests(manager);
manager.handleEvent(bundle,
new ServiceUnitStateData(ServiceUnitState.Owned, dstBroker, VERSION_ID_INIT), null);
assertEquals(inFlightUnloadRequests.size(), 1);

// Success with Init state.
manager.handleEvent(bundle,
new ServiceUnitStateData(ServiceUnitState.Init, dstBroker, VERSION_ID_INIT), null);
assertEquals(inFlightUnloadRequests.size(), 0);
counterExpected.update(SplitDecision.Label.Success, Sessions);
assertEquals(counter.toMetrics(null).toString(),
counterExpected.toMetrics(null).toString());
future.get();

// Success with Owned state.
future = manager.waitAsync(CompletableFuture.completedFuture(null),
bundle, decision, 5, TimeUnit.SECONDS);
inFlightUnloadRequests = getinFlightUnloadRequests(manager);
assertEquals(inFlightUnloadRequests.size(), 1);
manager.handleEvent(bundle,
new ServiceUnitStateData(ServiceUnitState.Owned, dstBroker, VERSION_ID_INIT), null);
assertEquals(inFlightUnloadRequests.size(), 0);
counterExpected.update(SplitDecision.Label.Success, Sessions);
assertEquals(counter.toMetrics(null).toString(),
counterExpected.toMetrics(null).toString());
future.get();
}