Denovo1998 commented on code in PR #26146:
URL: https://github.com/apache/pulsar/pull/26146#discussion_r3565550993
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java:
##########
@@ -280,6 +281,146 @@ public void channelOwnerTest() throws Exception {
}
}
+ @Test(priority = 1)
+ public void testCompletedGetOwnerRequestDoesNotRemoveNewRequest() throws
Exception {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl)
channel1;
+ String serviceUnit = namespaceName + "/0x10000000_0x10000001";
+ var getOwnerRequests = channel.getOwnerRequests();
+ getOwnerRequests.remove(serviceUnit);
+ CompletableFuture<String> oldRequest =
channel.dedupeGetOwnerRequest(serviceUnit);
+ assertEquals(getOwnerRequests.get(serviceUnit), oldRequest);
+
+ CountDownLatch staleCallbackRunning = new CountDownLatch(1);
+ CountDownLatch releaseStaleCallback = new CountDownLatch(1);
+ oldRequest.whenComplete((__, ___) -> {
+ staleCallbackRunning.countDown();
+ try {
+ assertTrue(releaseStaleCallback.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ }
+ });
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> completeOldRequest = executor.submit(() ->
oldRequest.complete(brokerId1));
+ assertTrue(staleCallbackRunning.await(5, TimeUnit.SECONDS),
+ "The stale get-owner completion callback did not start");
+ assertTrue(getOwnerRequests.get(serviceUnit) == oldRequest,
+ "The stale cleanup must not run before the new request is
installed");
+
+ // Replace the map entry before releasing the old callback.
+ CompletableFuture<String> newRequest = new CompletableFuture<>();
+ getOwnerRequests.put(serviceUnit, newRequest);
+ releaseStaleCallback.countDown();
+ completeOldRequest.get(5, TimeUnit.SECONDS);
+
+ assertTrue(getOwnerRequests.get(serviceUnit) == newRequest,
+ "A stale get-owner cleanup must not remove a newer request
future");
+ } finally {
+ releaseStaleCallback.countDown();
+ executor.shutdownNow();
+ getOwnerRequests.remove(serviceUnit);
+ }
+ }
+
+ @Test(priority = 1)
+ public void testSkippedEventDoesNotRemoveNewGetOwnerRequest() throws
Exception {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl)
channel1;
+ String serviceUnit = namespaceName + "/0x10000002_0x10000003";
+ var getOwnerRequests = channel.getOwnerRequests();
+ CompletableFuture<String> oldRequest = new CompletableFuture<>();
+ try {
+ overrideTableView(channel, serviceUnit, new
ServiceUnitStateData(Owned, brokerId1, 1));
+ getOwnerRequests.put(serviceUnit, oldRequest);
+
+ CountDownLatch staleCallbackRunning = new CountDownLatch(1);
+ CountDownLatch releaseStaleCallback = new CountDownLatch(1);
+ oldRequest.whenComplete((__, ___) -> {
+ staleCallbackRunning.countDown();
+ try {
+ assertTrue(releaseStaleCallback.await(5,
TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ }
+ });
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> handleSkippedEvent = executor.submit(() ->
channel.handleSkippedEvent(serviceUnit));
+ assertTrue(staleCallbackRunning.await(5, TimeUnit.SECONDS),
+ "The stale skipped-event callback did not start");
+ assertTrue(getOwnerRequests.get(serviceUnit) == oldRequest,
+ "The skipped-event cleanup must not run before the new
request is installed");
+
+ // Replace the map entry before releasing the old callback.
+ CompletableFuture<String> newRequest = new
CompletableFuture<>();
+ getOwnerRequests.put(serviceUnit, newRequest);
+ releaseStaleCallback.countDown();
+ handleSkippedEvent.get(5, TimeUnit.SECONDS);
+
+ assertEquals(oldRequest.getNow(null), brokerId1);
+ assertTrue(getOwnerRequests.get(serviceUnit) == newRequest,
+ "A stale skipped-event cleanup must not remove a newer
request future");
+ } finally {
+ releaseStaleCallback.countDown();
+ executor.shutdownNow();
+ }
+ } finally {
+ getOwnerRequests.remove(serviceUnit);
+ oldRequest.cancel(false);
+ overrideTableView(channel, serviceUnit, null);
+ }
+ }
+
+ @Test(priority = 1)
+ public void testCompletedCleanupJobDoesNotRemoveNewCleanupJob() throws
Exception {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl)
channel1;
+ String broker = brokerId3;
+ var cleanupJobs = channel.getCleanupJobs();
+ cleanupJobs.remove(broker);
+ channel.scheduleCleanup(broker, 60L);
+ CompletableFuture<Void> oldJob = cleanupJobs.get(broker);
+ assertNotNull(oldJob);
+
+ CountDownLatch staleCallbackRunning = new CountDownLatch(1);
+ CountDownLatch releaseStaleCallback = new CountDownLatch(1);
+ oldJob.whenComplete((__, ___) -> {
Review Comment:
This has the same dependent-action ordering assumption as
testCompletedGetOwnerRequestDoesNotRemoveNewRequest: the blocking callback
registered here must run before the cleanup callback registered by
scheduleCleanup, but CompletableFuture does not guarantee that ordering. We can
avoid the latches and executor by replacing the map entry with newJob before
calling oldJob.complete(null). The old implementation will then
deterministically remove newJob, while the conditional-remove implementation
will preserve it.
##########
pulsar-broker/src/test/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelTest.java:
##########
@@ -280,6 +281,146 @@ public void channelOwnerTest() throws Exception {
}
}
+ @Test(priority = 1)
+ public void testCompletedGetOwnerRequestDoesNotRemoveNewRequest() throws
Exception {
+ ServiceUnitStateChannelImpl channel = (ServiceUnitStateChannelImpl)
channel1;
+ String serviceUnit = namespaceName + "/0x10000000_0x10000001";
+ var getOwnerRequests = channel.getOwnerRequests();
+ getOwnerRequests.remove(serviceUnit);
+ CompletableFuture<String> oldRequest =
channel.dedupeGetOwnerRequest(serviceUnit);
+ assertEquals(getOwnerRequests.get(serviceUnit), oldRequest);
+
+ CountDownLatch staleCallbackRunning = new CountDownLatch(1);
+ CountDownLatch releaseStaleCallback = new CountDownLatch(1);
+ oldRequest.whenComplete((__, ___) -> {
Review Comment:
This test assumes that the whenComplete registered here runs before the
cleanup callback registered by dedupeGetOwnerRequest. CompletableFuture does
not specify the execution order of dependent actions. If the production cleanup
runs first, the patched implementation can legitimately remove the old mapping
before staleCallbackRunning is released, causing the assertion below to fail.
Could we make this deterministic by installing newRequest before completing
oldRequest, then completing the old request and asserting that the new mapping
remains? That would reproduce the stale cleanup without relying on callback
ordering or an executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
