sandeep-mst commented on code in PR #25067:
URL: https://github.com/apache/pulsar/pull/25067#discussion_r2611415894
##########
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
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]
