Copilot commented on code in PR #25067:
URL: https://github.com/apache/pulsar/pull/25067#discussion_r2611388147
##########
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:
The comment contains a spelling inconsistency: "rackinfo" should be "rack
info" (two words) to match the usage throughout the rest of the codebase and
comments.
```suggestion
// Rack info update is delayed but because of new callback the rack
info on ensemble policy should be updated.
```
##########
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:
The comment states "Rackaware" but should be "RackAware" to match the
standard capitalization convention used in the codebase (e.g.,
"RackawareEnsemblePlacementPolicy" in code vs "rack-aware" in general text).
Consider updating to "RackAware ensemble policy listener" or keeping it
lowercase as "rack-aware ensemble policy listener" for consistency with
technical documentation style.
```suggestion
// watcher.processWritableBookiesChanged runs FIRST triggering
RackAware ensemble policy listener → incorrect
```
##########
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:
This commented-out code should either be removed or uncommented if it's
needed for the test. Leaving commented code in production reduces code
cleanliness and can cause confusion about whether it was intentionally disabled
or forgotten.
If this assertion is not needed, please remove lines 500-503. If it is
needed for verification, uncomment it.
```suggestion
List<String> resolved =
mapping.resolve(Lists.newArrayList(bookie1.getHostName()));
assertEquals(resolved.get(0), "/rack0",
"Expected mapping to have /rack0 after update before watcher
ran");
```
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMapping.java:
##########
@@ -169,6 +172,13 @@ private void watchAvailableBookies() {
}
}
+ private Void processRackUpdate(BookiesRackConfiguration racks,
List<BookieId> bookieAddressListLastTime) {
+ updateRacksWithHost(racks);
+ // Adding callback after rackInfo is updated by change in writable
bookies.
+ rackChangeListenerCallback(bookieAddressListLastTime);
+ return null;
+ }
Review Comment:
The processRackUpdate method accesses the bookieAddressListLastTime field
without synchronization, which could lead to race conditions. The
bookieAddressListLastTime field is accessed and modified in multiple places
(setConf, handleUpdates), all within synchronized blocks. However, this method
is called from watchAvailableBookies callback which runs asynchronously.
Consider making this method synchronized or ensuring it's called within a
synchronized block to maintain thread safety when accessing shared mutable
state.
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMapping.java:
##########
@@ -169,6 +172,13 @@ private void watchAvailableBookies() {
}
}
+ private Void processRackUpdate(BookiesRackConfiguration racks,
List<BookieId> bookieAddressListLastTime) {
+ updateRacksWithHost(racks);
+ // Adding callback after rackInfo is updated by change in writable
bookies.
Review Comment:
The comment "Adding callback after rackInfo is updated by change in writable
bookies" could be more precise. Consider clarifying that this ensures the
ensemble placement policy is notified only after the rack mapping has been
updated, which fixes the race condition where the policy might be notified
before the mapping is complete.
For example: "Notify ensemble placement policy after rack info is updated to
ensure consistent state"
```suggestion
// Notify ensemble placement policy after rack info is updated to
ensure consistent state.
// This ordering avoids a race condition where the policy might be
notified before the mapping is complete.
```
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/bookie/rackawareness/BookieRackAffinityMapping.java:
##########
@@ -274,12 +284,16 @@ private void handleUpdates(Notification n) {
bookieIdSet.addAll(bookieAddressListLastTime);
bookieAddressListLastTime = bookieAddressList;
}
- if (rackawarePolicy != null) {
- rackawarePolicy.onBookieRackChange(new
ArrayList<>(bookieIdSet));
- }
+ rackChangeListenerCallback(bookieIdSet);
});
}
+ private void rackChangeListenerCallback(Collection<BookieId> bookieIdSet) {
+ if (rackawarePolicy != null) {
+ rackawarePolicy.onBookieRackChange(new ArrayList<>(bookieIdSet));
+ }
+ }
Review Comment:
The rackawarePolicy field is accessed without synchronization in the
rackChangeListenerCallback method (line 292-293), while it can be written by
registerRackChangeListener (line 299). Although registerRackChangeListener is
typically called during initialization, there's no guarantee about the
visibility of the updated rackawarePolicy value across threads without proper
synchronization or volatile declaration.
Consider either making rackawarePolicy volatile, or ensuring
rackChangeListenerCallback accesses it within a synchronized block, or
documenting that registerRackChangeListener must be called before any
concurrent operations begin.
--
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]
