apurtell commented on code in PR #2547:
URL: https://github.com/apache/phoenix/pull/2547#discussion_r3501830439
##########
phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java:
##########
@@ -866,214 +885,203 @@ private void syncZKToSystemTable() {
}
}
- private void maybeInitializePeerPathChildrenCache() throws IOException {
- // There is an edge case when the cache is not initialized yet, but we get
CHILD_ADDED event
- // so we need to get the record from ZK.
- HAGroupStoreRecord currentHAGroupStoreRecord =
- phoenixHaAdmin.getHAGroupStoreRecordInZooKeeper(haGroupName).getLeft();
- if (currentHAGroupStoreRecord == null) {
- LOGGER.error(
- "Current HAGroupStoreRecord is null," + "skipping peer path children
cache initialization");
- return;
- }
- String peerZKUrl = currentHAGroupStoreRecord.getPeerZKUrl();
- if (StringUtils.isNotBlank(peerZKUrl)) {
- try {
- // Setup peer connection if needed (first time or ZK Url changed)
- if (
- peerPathChildrenCache == null || peerPhoenixHaAdmin != null
- && !StringUtils.equals(peerZKUrl, peerPhoenixHaAdmin.getZkUrl())
- ) {
- // Clean up existing peer connection if it exists
- closePeerConnection();
- // Setup new peer connection
- this.peerPhoenixHaAdmin =
- new PhoenixHAAdmin(peerZKUrl, conf,
ZK_CONSISTENT_HA_GROUP_RECORD_NAMESPACE);
- // Create new PeerPathChildrenCache
- this.peerPathChildrenCache =
initializePathChildrenCache(peerPhoenixHaAdmin,
- this.peerCustomPathChildrenCacheListener, ClusterType.PEER);
- }
- } catch (Exception e) {
- closePeerConnection();
- LOGGER.error("Unable to initialize PeerPathChildrenCache for
HAGroupStoreClient", e);
- // Don't think we should mark HAGroupStoreClient as unhealthy if
- // peerCache is unhealthy, if needed we can introduce a config to
control behavior.
- }
- } else {
- // Close Peer Cache for this HAGroupName if currentClusterRecord is null
- // or peerZKUrl is blank
- closePeerConnection();
- LOGGER.error("Not initializing PeerPathChildrenCache for
HAGroupStoreClient "
- + "with HAGroupName {} as peerZKUrl is blank", haGroupName);
- }
- }
-
- private PathChildrenCache initializePathChildrenCache(PhoenixHAAdmin admin,
- PathChildrenCacheListener customListener, ClusterType cacheType) {
- LOGGER.info("Initializing {} PathChildrenCache with URL {}", cacheType,
admin.getZkUrl());
- PathChildrenCache newPathChildrenCache = null;
- try {
- newPathChildrenCache =
- new PathChildrenCache(admin.getCurator(), ZKPaths.PATH_SEPARATOR,
true);
- final CountDownLatch latch = new CountDownLatch(1);
- newPathChildrenCache.getListenable().addListener(
- customListener != null ? customListener : createCacheListener(latch,
cacheType));
-
newPathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
- boolean initialized =
-
latch.await(conf.getLong(PHOENIX_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS,
- DEFAULT_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS),
TimeUnit.MILLISECONDS);
- if (!initialized && customListener == null) {
- newPathChildrenCache.close();
- return null;
- }
- return newPathChildrenCache;
- } catch (Exception e) {
- if (newPathChildrenCache != null) {
- try {
- newPathChildrenCache.close();
- } catch (IOException ioe) {
- LOGGER.error("Failed to close {} PathChildrenCache with ZKUrl",
cacheType, ioe);
- }
- }
- LOGGER.error("Failed to initialize {} PathChildrenCache", cacheType, e);
- return null;
- }
- }
-
- private PathChildrenCacheListener createCacheListener(CountDownLatch latch,
- ClusterType cacheType) {
+ /**
+ * Listener for the local cache: tracks local state transitions (suppressing
STANDBY while
+ * degraded), keeps the peer watcher pointed at the current peer ZK, drives
the legacy CRR sync,
+ * and toggles health on connection loss/reconnect. The initial-load latch
is handled by
+ * {@link HAGroupStoreCacheUtil#startCache}.
+ */
+ private PathChildrenCacheListener localCacheListener() {
return (client, event) -> {
- final ChildData childData = event.getData();
- Pair<HAGroupStoreRecord, Stat> eventRecordAndStat =
- extractHAGroupStoreRecordOrNull(childData);
- HAGroupStoreRecord eventRecord = eventRecordAndStat.getLeft();
- Stat eventStat = eventRecordAndStat.getRight();
- if (eventRecord != null && !Objects.equals(eventRecord.getHaGroupName(),
haGroupName)) {
+ Pair<HAGroupStoreRecord, Stat> recordAndStat =
+ HAGroupStoreCacheUtil.recordAndStat(event.getData());
+ HAGroupStoreRecord record = recordAndStat.getLeft();
+ if (record != null && !Objects.equals(record.getHaGroupName(),
haGroupName)) {
return;
}
- LOGGER.info(
- "HAGroupStoreClient Cache {} received event {} type {} at {} with
ZKUrl {} and "
- + "PeerZKUrl {} for haGroupName {}",
- cacheType, eventRecord, event.getType(), System.currentTimeMillis(),
- phoenixHaAdmin.getZkUrl(),
- peerPhoenixHaAdmin != null ? peerPhoenixHaAdmin.getZkUrl() :
"peerPhoenixHaAdmin is null",
- haGroupName);
switch (event.getType()) {
case CHILD_ADDED:
case CHILD_UPDATED:
- if (eventRecord != null &&
Objects.equals(eventRecord.getHaGroupName(), haGroupName)) {
- handleStateChange(eventRecord, eventStat, cacheType);
- // Reinitialize peer path children cache if peer url is added or
updated.
- if (cacheType == ClusterType.LOCAL) {
- maybeInitializePeerPathChildrenCache();
- }
- // Offload the legacy CRR sync (it does ZK + JDBC I/O) so we don't
block
- // Curator's per-namespace event dispatcher.
- ScheduledExecutorService syncExec = legacyCrrSyncExecutor;
- if (syncExec != null) {
- try {
- syncExec.execute(this::syncLegacyCRRIfRoleChanged);
- } catch (RejectedExecutionException ree) {
- // Executor already shutting down (close() race); drop
silently.
- LOGGER.debug("Legacy CRR sync skipped for HA group {}:
executor shut down",
- haGroupName);
- }
+ if (record != null) {
+ handleLocalStateChange(record, recordAndStat.getRight());
+ // Reconcile only when the peer ZK URL changed, off the Curator
dispatcher so a slow or
+ // unreachable peer never blocks local event processing.
+ String peerZKUrl = record.getPeerZKUrl();
+ if (!Objects.equals(peerZKUrl, lastConfiguredPeerZKUrl)) {
+ lastConfiguredPeerZKUrl = peerZKUrl;
+ peerWatcher.reconfigureAsync(peerZKUrl);
}
+ offloadLegacyCrrSync();
}
break;
- case CHILD_REMOVED:
- // No-op: the legacy /phoenix/ha znode is never deleted by this
client.
- break;
- case INITIALIZED:
- latch.countDown();
- break;
case CONNECTION_LOST:
case CONNECTION_SUSPENDED:
- if (ClusterType.LOCAL.equals(cacheType)) {
- isHealthy = false;
- }
- LOGGER.warn("{} HAGroupStoreClient cache connection lost/suspended",
cacheType);
+ isHealthy = false;
+ LOGGER.warn("LOCAL HAGroupStoreClient cache connection
lost/suspended for HA group {}",
+ haGroupName);
break;
case CONNECTION_RECONNECTED:
- if (ClusterType.LOCAL.equals(cacheType)) {
- isHealthy = true;
- }
- LOGGER.info("{} HAGroupStoreClient cache connection reconnected",
cacheType);
+ isHealthy = true;
+ LOGGER.info("LOCAL HAGroupStoreClient cache connection reconnected
for HA group {}",
+ haGroupName);
break;
default:
- LOGGER.warn("Unexpected {} event type {}, complete event {}",
cacheType, event.getType(),
- event);
+ break;
}
};
}
- private Pair<HAGroupStoreRecord, Stat>
- fetchCacheRecordAndPopulateZKIfNeeded(PathChildrenCache cache, ClusterType
cacheType) {
- if (cache == null) {
- LOGGER.warn("{} HAGroupStoreClient cache is null, returning null",
cacheType);
+ /**
+ * Read the local record from the cache; if absent, rebuild once from the
system table (the znode
+ * may have been deleted) and re-read. Returns (null, null) when still
absent.
+ */
+ private Pair<HAGroupStoreRecord, Stat>
fetchLocalRecordAndPopulateZKIfNeeded() {
+ if (pathChildrenCache == null) {
+ LOGGER.warn("LOCAL HAGroupStoreClient cache is null for HA group {},
returning null",
+ haGroupName);
return Pair.of(null, null);
}
String targetPath = toPath(this.haGroupName);
- // Try to get record from current cache data
- Pair<HAGroupStoreRecord, Stat> result = extractRecordAndStat(cache,
targetPath, cacheType);
+ Pair<HAGroupStoreRecord, Stat> result =
+ HAGroupStoreCacheUtil.recordAndStatAt(pathChildrenCache, targetPath);
if (result.getLeft() != null) {
return result;
}
-
- if (cacheType.equals(ClusterType.PEER)) {
- return Pair.of(null, null);
- }
- // If no record found, try to rebuild and fetch again
- LOGGER.info("No record found at path {} for {} cluster, trying to
initialize ZNode "
- + "from System Table in case it might have been deleted", targetPath,
cacheType);
+ LOGGER.info("No record found at path {} for LOCAL cluster, trying to
initialize ZNode from "
+ + "System Table in case it might have been deleted", targetPath);
try {
rebuild();
- return extractRecordAndStat(cache, targetPath, cacheType);
+ return HAGroupStoreCacheUtil.recordAndStatAt(pathChildrenCache,
targetPath);
} catch (Exception e) {
- LOGGER.error(
- "Failed to initialize ZNode from System Table, giving up " + "and
returning null", e);
+ LOGGER.error("Failed to initialize ZNode from System Table, giving up
and returning null", e);
return Pair.of(null, null);
}
}
- private Pair<HAGroupStoreRecord, Stat>
extractRecordAndStat(PathChildrenCache cache,
- String targetPath, ClusterType cacheType) {
- ChildData childData = cache.getCurrentData(targetPath);
- if (childData != null) {
- Pair<HAGroupStoreRecord, Stat> recordAndStat =
extractHAGroupStoreRecordOrNull(childData);
- LOGGER.debug("Built {} cluster record: {}", cacheType,
recordAndStat.getLeft());
- return recordAndStat;
+ /**
+ * The effective local HA record, or {@code null} when no local record
exists. Identical to
+ * {@link #getHAGroupStoreRecord()} except that, while this RegionServer
cannot see the peer, a
+ * local STANDBY is reported as DEGRADED_STANDBY. The DEGRADED_STANDBY
overlay is in-memory only
+ * and is never written to the shared HA record; peer connectivity is never
exposed directly. The
+ * overlay is applied only on top of a successfully read local record, so
this requires the LOCAL
+ * client to be healthy: a lost LOCAL connection is bounded (Curator
reconnects, or the
+ * RegionServer is aborted on prolonged ZK loss), so no stale-record
fallback is provided here.
+ * @throws IOException if the LOCAL client is not healthy (same contract as
+ * {@link #getHAGroupStoreRecord()})
+ */
+ public HAGroupStoreRecord getEffectiveHAGroupStoreRecord() throws
IOException {
Review Comment:
getHAGroupStoreRecord() can call rebuild(), which calls
getHAGroupStoreRecordInZooKeeper, then a query from SYSTEM.HA_GROUP, then
createHAGroupStoreRecordInZooKeeper, all while synchronized on
localDegradedStandbyLock. The peer watcher's onPeerBlind and onPeerVisible
listeners also need to acquire localDegradedStandbyLock so a slow rebuild will
pile up peer visibility callbacks behind it.
Consider snapshotting under the lock and rebuilding outside it.
--
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]
