Jackie-Jiang commented on code in PR #20107:
URL: https://github.com/apache/pinot/pull/20107#discussion_r4430025070
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java:
##########
@@ -103,23 +115,62 @@ public synchronized void
onInstanceConfigChange(List<InstanceConfig> instanceCon
}
}
- /// Full cluster scan: used on INIT (first callback) and when instances are
added/removed.
+ /// Scans on INIT and child changes, skipping membership changes confined to
non-server instances.
///
/// It records the new address of every server before it retires any old
one. Otherwise, when one server left an
/// address and a server read later in the scan took it over, the scan would
shut down the channel the second server
/// uses.
- private void handleFullScan() {
+ private void handleFullScan(NotificationContext context) {
Review Comment:
Optional cleanup: rename this to handleMembershipChange, since it may skip
scanning, and extract the actual server reconciliation into
scanServerConfigs(instances), returning whether all configs were read
successfully. Keeping the metadata read, skip decision, and snapshot
publication together would make the cache lifecycle easier to follow.
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandler.java:
##########
@@ -68,6 +74,12 @@ public class ServerMailboxChannelRefreshHandler implements
InstanceConfigChangeL
private final String _selfInstanceId;
private final MailboxService _mailboxService;
+ // A coherent participant snapshot, invalidated whenever a membership or
instance config read fails.
+ private Set<String> _instances = Set.of();
+ private long _instancesParentCzxid;
+ private int _instancesParentCversion;
+ private boolean _instancesSnapshotValid;
Review Comment:
Optional cleanup: represent the cache as one nullable immutable
MembershipSnapshot containing the names, parent czxid, and cversion. Null can
mean invalid. Capture the previous snapshot and clear the field before reading,
then publish a replacement only after a proven skip or successful coherent
scan. This makes invalidation one assignment and avoids coordinating four
fields.
##########
pinot-server/src/test/java/org/apache/pinot/server/starter/helix/ServerMailboxChannelRefreshHandlerTest.java:
##########
@@ -206,12 +223,163 @@ public void testInitCallbackDoesFullScanAndTracksState()
{
verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
}
+ @Test
+ public void testMinionMembershipChangesDoNotReadServerConfigs() {
+ String minionId = "Minion_minionhost_8000";
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 10, 1);
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+
+ stubParticipantSnapshot(List.of(minionId, OTHER_SERVER_ID,
SELF_INSTANCE_ID), 11, 1);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID), 12, 1);
+ _handler.onInstanceConfigChange(List.of(), createChildChangeContext());
+ verify(_helixAdmin, times(1)).getInstanceConfig(CLUSTER_NAME,
OTHER_SERVER_ID);
+ verify(_mailboxService, never()).refreshChannel(any(), anyInt());
+
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, false));
+ _handler.onInstanceConfigChange(List.of(),
createCallbackContextForInstance(OTHER_SERVER_ID));
+ verify(_mailboxService).refreshChannel(OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT);
+ }
+
+ @Test
+ public void testServerMembershipChangesStillScanConfigs() {
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID));
+ when(_helixAdmin.getInstanceConfig(CLUSTER_NAME, OTHER_SERVER_ID))
+ .thenReturn(createServerConfig(OTHER_SERVER_ID, OTHER_SERVER_HOST,
OTHER_SERVER_MAILBOX_PORT, true));
+ _handler.onInstanceConfigChange(List.of(), createInitContext());
+ stubParticipantSnapshot(List.of(SELF_INSTANCE_ID, OTHER_SERVER_ID));
Review Comment:
Optional test cleanup: use explicit cversion values such as 10 then 11 in
the server add/remove tests. The default helper keeps cversion at 0 across the
membership change, so the version mismatch already forces a scan. A realistic
increment would ensure these tests also exercise the server-type guard. The
same applies to testRetiresChannelOfServerThatLeftTheCluster.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
