void-ptr974 commented on code in PR #26146:
URL: https://github.com/apache/pulsar/pull/26146#discussion_r3566280698
##########
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:
Updated in 12d970b138b. The test now installs the replacement request before
completing the old one, so it deterministically exercises the stale cleanup
without relying on dependent-action ordering or an executor.
##########
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:
Updated in 12d970b138b. This test now replaces the cleanup job before
completing the old job, making the stale removal deterministic and eliminating
the latches and 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]
