sandeep-mst commented on code in PR #25067:
URL: https://github.com/apache/pulsar/pull/25067#discussion_r2611414825
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMappingTest.java:
##########
@@ -415,4 +415,111 @@ public void testNoDeadlockWithRackawarePolicy() throws
Exception {
assertTrue(count.await(3, TimeUnit.SECONDS));
}
+
+ @Test
+ public void testZKEventListenersOrdering() throws Exception {
+ @Cleanup
+ PulsarRegistrationClient pulsarRegistrationClient =
+ new PulsarRegistrationClient(store, "/ledgers");
+ DefaultBookieAddressResolver defaultBookieAddressResolver =
+ new DefaultBookieAddressResolver(pulsarRegistrationClient);
+ // Create and configure the mapping
+ BookieRackAffinityMapping mapping = new BookieRackAffinityMapping();
+ ClientConfiguration bkClientConf = new ClientConfiguration();
+
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE,
store);
+ mapping.setBookieAddressResolver(defaultBookieAddressResolver);
+ mapping.setConf(bkClientConf);
+
+ // Create RackawareEnsemblePlacementPolicy and initialize it
+ @Cleanup("stop")
+ HashedWheelTimer timer = getTestHashedWheelTimer(bkClientConf);
+ RackawareEnsemblePlacementPolicy repp = new
RackawareEnsemblePlacementPolicy();
+ repp.initialize(bkClientConf, Optional.of(mapping), timer,
+ DISABLE_ALL, NullStatsLogger.INSTANCE,
defaultBookieAddressResolver);
+ mapping.registerRackChangeListener(repp);
+
+ // Create a BookieWatcherImpl instance via reflection
+ Class<?> watcherClazz =
Class.forName("org.apache.bookkeeper.client.BookieWatcherImpl");
+ Constructor<?> constructor = watcherClazz.getDeclaredConstructor(
+ ClientConfiguration.class,
+ EnsemblePlacementPolicy.class,
+ RegistrationClient.class,
+ BookieAddressResolver.class,
+ StatsLogger.class);
+ constructor.setAccessible(true);
+ Object watcher = constructor.newInstance(
+ bkClientConf,
+ repp,
+ pulsarRegistrationClient,
+ defaultBookieAddressResolver,
+ NullStatsLogger.INSTANCE
+ );
+ Method initMethod =
watcherClazz.getDeclaredMethod("initialBlockingBookieRead");
+ initMethod.setAccessible(true);
+ initMethod.invoke(watcher);
+
+ // Prepare a BookiesRackConfiguration that maps bookie1 -> /rack0
+ BookieInfo bi = BookieInfo.builder().rack("/rack0").build();
+ BookiesRackConfiguration racks = new BookiesRackConfiguration();
+ racks.updateBookie("group1", bookie1.toString(), bi);
+
+ // Create a mock cache for racks /bookies
+ MetadataCache<BookiesRackConfiguration> mockCache =
mock(MetadataCache.class);
+ Field f =
BookieRackAffinityMapping.class.getDeclaredField("bookieMappingCache");
+ f.setAccessible(true);
+ f.set(mapping, mockCache);
+ when(mockCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH))
+
.thenReturn(CompletableFuture.completedFuture(Optional.of(racks)));
+
+ // Inject the writable bookie into PulsarRegistrationClient
+ Field writableField =
PulsarRegistrationClient.class.getDeclaredField("writableBookieInfo");
+ writableField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<BookieId, Versioned<BookieServiceInfo>> writableBookieInfo =
+ (Map<BookieId, Versioned<BookieServiceInfo>>)
writableField.get(pulsarRegistrationClient);
+ writableBookieInfo.put(
+ bookie1.toBookieId(),
+ new
Versioned<>(BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookie1.toString()),
Version.NEW)
+ );
+
+ // watcher.processWritableBookiesChanged runs FIRST triggering
Rackaware ensemble policy listener → incorrect
+ // ordering
+ Method procMethod =
+
watcherClazz.getDeclaredMethod("processWritableBookiesChanged",
java.util.Set.class);
+ procMethod.setAccessible(true);
+ Set<BookieId> ids = new HashSet<>();
+ ids.add(bookie1.toBookieId());
+ procMethod.invoke(watcher, ids);
+
+ // BookieRackAffinityMapping rack mapping update runs SECOND → delayed
rack info
+ Method processRackUpdateMethod =
BookieRackAffinityMapping.class.getDeclaredMethod("processRackUpdate",
+ BookiesRackConfiguration.class, List.class);
+ processRackUpdateMethod.setAccessible(true);
+ processRackUpdateMethod.invoke(mapping, racks,
List.of(bookie1.toBookieId()));
+
+ // mapping.resolve now has correct rack (mapping is updated)
+// List<String> resolved =
mapping.resolve(Lists.newArrayList(bookie1.getHostName()));
+// assertEquals(resolved.get(0), "/rack0",
+// "Expected mapping to have /rack0 after update before watcher
ran");
+
+ // -------------------
+ // NOW CHECK REPP INTERNAL STATE
+ // -------------------
+ // BookieNode.getNetworkLocation()
+ Class<?> clazz1 =
Class.forName("org.apache.bookkeeper.client.TopologyAwareEnsemblePlacementPolicy");
+ Field field1 = clazz1.getDeclaredField("knownBookies");
+ field1.setAccessible(true);
+ Map<BookieId, BookieNode> knownBookies = (Map<BookieId, BookieNode>)
field1.get(repp);
+ BookieNode bn = knownBookies.get(bookie1.toBookieId());
+ // Rack info update is delayed but because of new callback the
rackinfo on ensemble policy should be updated.
Review Comment:
done
##########
pulsar-broker-common/src/test/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMappingTest.java:
##########
@@ -415,4 +415,111 @@ public void testNoDeadlockWithRackawarePolicy() throws
Exception {
assertTrue(count.await(3, TimeUnit.SECONDS));
}
+
+ @Test
+ public void testZKEventListenersOrdering() throws Exception {
+ @Cleanup
+ PulsarRegistrationClient pulsarRegistrationClient =
+ new PulsarRegistrationClient(store, "/ledgers");
+ DefaultBookieAddressResolver defaultBookieAddressResolver =
+ new DefaultBookieAddressResolver(pulsarRegistrationClient);
+ // Create and configure the mapping
+ BookieRackAffinityMapping mapping = new BookieRackAffinityMapping();
+ ClientConfiguration bkClientConf = new ClientConfiguration();
+
bkClientConf.setProperty(BookieRackAffinityMapping.METADATA_STORE_INSTANCE,
store);
+ mapping.setBookieAddressResolver(defaultBookieAddressResolver);
+ mapping.setConf(bkClientConf);
+
+ // Create RackawareEnsemblePlacementPolicy and initialize it
+ @Cleanup("stop")
+ HashedWheelTimer timer = getTestHashedWheelTimer(bkClientConf);
+ RackawareEnsemblePlacementPolicy repp = new
RackawareEnsemblePlacementPolicy();
+ repp.initialize(bkClientConf, Optional.of(mapping), timer,
+ DISABLE_ALL, NullStatsLogger.INSTANCE,
defaultBookieAddressResolver);
+ mapping.registerRackChangeListener(repp);
+
+ // Create a BookieWatcherImpl instance via reflection
+ Class<?> watcherClazz =
Class.forName("org.apache.bookkeeper.client.BookieWatcherImpl");
+ Constructor<?> constructor = watcherClazz.getDeclaredConstructor(
+ ClientConfiguration.class,
+ EnsemblePlacementPolicy.class,
+ RegistrationClient.class,
+ BookieAddressResolver.class,
+ StatsLogger.class);
+ constructor.setAccessible(true);
+ Object watcher = constructor.newInstance(
+ bkClientConf,
+ repp,
+ pulsarRegistrationClient,
+ defaultBookieAddressResolver,
+ NullStatsLogger.INSTANCE
+ );
+ Method initMethod =
watcherClazz.getDeclaredMethod("initialBlockingBookieRead");
+ initMethod.setAccessible(true);
+ initMethod.invoke(watcher);
+
+ // Prepare a BookiesRackConfiguration that maps bookie1 -> /rack0
+ BookieInfo bi = BookieInfo.builder().rack("/rack0").build();
+ BookiesRackConfiguration racks = new BookiesRackConfiguration();
+ racks.updateBookie("group1", bookie1.toString(), bi);
+
+ // Create a mock cache for racks /bookies
+ MetadataCache<BookiesRackConfiguration> mockCache =
mock(MetadataCache.class);
+ Field f =
BookieRackAffinityMapping.class.getDeclaredField("bookieMappingCache");
+ f.setAccessible(true);
+ f.set(mapping, mockCache);
+ when(mockCache.get(BookieRackAffinityMapping.BOOKIE_INFO_ROOT_PATH))
+
.thenReturn(CompletableFuture.completedFuture(Optional.of(racks)));
+
+ // Inject the writable bookie into PulsarRegistrationClient
+ Field writableField =
PulsarRegistrationClient.class.getDeclaredField("writableBookieInfo");
+ writableField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<BookieId, Versioned<BookieServiceInfo>> writableBookieInfo =
+ (Map<BookieId, Versioned<BookieServiceInfo>>)
writableField.get(pulsarRegistrationClient);
+ writableBookieInfo.put(
+ bookie1.toBookieId(),
+ new
Versioned<>(BookieServiceInfoUtils.buildLegacyBookieServiceInfo(bookie1.toString()),
Version.NEW)
+ );
+
+ // watcher.processWritableBookiesChanged runs FIRST triggering
Rackaware ensemble policy listener → incorrect
+ // ordering
+ Method procMethod =
+
watcherClazz.getDeclaredMethod("processWritableBookiesChanged",
java.util.Set.class);
+ procMethod.setAccessible(true);
+ Set<BookieId> ids = new HashSet<>();
+ ids.add(bookie1.toBookieId());
+ procMethod.invoke(watcher, ids);
+
+ // BookieRackAffinityMapping rack mapping update runs SECOND → delayed
rack info
+ Method processRackUpdateMethod =
BookieRackAffinityMapping.class.getDeclaredMethod("processRackUpdate",
+ BookiesRackConfiguration.class, List.class);
+ processRackUpdateMethod.setAccessible(true);
+ processRackUpdateMethod.invoke(mapping, racks,
List.of(bookie1.toBookieId()));
+
+ // mapping.resolve now has correct rack (mapping is updated)
+// List<String> resolved =
mapping.resolve(Lists.newArrayList(bookie1.getHostName()));
+// assertEquals(resolved.get(0), "/rack0",
+// "Expected mapping to have /rack0 after update before watcher
ran");
Review Comment:
done
--
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]
