Copilot commented on code in PR #2547: URL: https://github.com/apache/phoenix/pull/2547#discussion_r3500664753
########## phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PeerClusterWatcher.java: ########## @@ -0,0 +1,428 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.phoenix.jdbc; + +import static org.apache.phoenix.jdbc.PhoenixHAAdmin.toPath; +import static org.apache.phoenix.query.QueryServices.HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS; +import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Objects; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.curator.framework.recipes.cache.PathChildrenCache; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; +import org.apache.hadoop.conf.Configuration; +import org.apache.zookeeper.data.Stat; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting; +import org.apache.phoenix.thirdparty.com.google.common.util.concurrent.MoreExecutors; + +/** + * Watches one peer cluster's {@link HAGroupStoreRecord} over the peer's ZooKeeper, for a single HA + * group on one RegionServer. Owns the peer cache + admin and a background retry, and reports peer + * state changes (de-duplicated by znode version, with one forced redelivery after a reconnect) and + * visible<->blind transitions. Thread-safe; the (possibly blocking) cache build runs on the + * caller's thread for the initial {@link #reconfigure} and on the retry executor afterwards, never + * holding {@link #stateLock}. Listener callbacks fire outside the lock. + */ +final class PeerClusterWatcher implements Closeable { + + /** + * Sink for what the watcher observes; implemented by {@link HAGroupStoreClient}. Callbacks run + * while the watcher holds {@code transitionLock} (outside {@code stateLock}), so implementations + * must not re-enter the watcher (e.g. {@code reconfigure} / {@code close}) and should offload + * blocking work. + */ + interface PeerStateListener { + /** + * Current peer HA record. May be redelivered once after reconnect even if the znode version did + * not change; consumers must tolerate duplicate same-state delivery. + */ + void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat); + + /** Peer connectivity is visible again; this is not a peer HA state transition. */ + void onPeerVisible(); + + /** Peer connectivity is unavailable; this is not a peer HA state transition. */ + void onPeerBlind(); + } + + private enum Visibility { + UNKNOWN, + VISIBLE, + BLIND + } + + private static final Logger LOGGER = LoggerFactory.getLogger(PeerClusterWatcher.class); + private static final long RETRY_WARN_EVERY_N_ATTEMPTS = 10L; + + private final Configuration conf; + private final String haGroupName; + private final String namespace; + private final PeerStateListener listener; + private final long initTimeoutMs; + private final long retryIntervalSec; + + // Serializes a whole reconcile (close/build/publish) so the constructor's synchronous reconcile + // and the retry executor never build concurrently. Held only by ensureConnection; + // stateLock is taken briefly inside it for field access (ordering is always + // reconcileLock -> stateLock). + private final Object reconcileLock = new Object(); + private final Object stateLock = new Object(); + // Serializes a visibility transition with its notification so concurrent transitions (peer event + // thread vs reconcile/retry executor) cannot reorder the notifications they deliver. + private final Object transitionLock = new Object(); + private String peerZkUrl; // desired peer; blank = none + private PhoenixHAAdmin admin; + private PathChildrenCache cache; + private int lastDeliveredVersion = -1; + private long peerCacheRetryAttempts = 0L; + private volatile boolean watcherClosed = false; + private ScheduledExecutorService retryExecutor; + private boolean retryScheduled = false; + private volatile Visibility visibility = Visibility.UNKNOWN; + + PeerClusterWatcher(Configuration conf, String haGroupName, String namespace, + PeerStateListener listener) { + this.conf = conf; + this.haGroupName = haGroupName; + this.namespace = namespace; + this.listener = listener; + this.initTimeoutMs = + conf.getLong(HAGroupStoreClient.PHOENIX_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS, + HAGroupStoreClient.DEFAULT_HA_GROUP_STORE_CLIENT_INITIALIZATION_TIMEOUT_MS); + this.retryIntervalSec = conf.getLong(HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS, + DEFAULT_HA_GROUP_STORE_PEER_CACHE_RETRY_INTERVAL_SECONDS); + // Create the executor up front (no worker thread starts until the first task is submitted) so + // reconfigureAsync can run off the caller's thread. The periodic retry is scheduled lazily on + // the first reconfigure with a peer; a watcher that is never configured never starts a thread. + this.retryExecutor = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "PeerClusterWatcher-" + haGroupName); + t.setDaemon(true); + return t; + }); + } + + /** Set/change/clear the peer and reconcile the connection synchronously. */ + void reconfigure(String url) { + synchronized (stateLock) { + if (watcherClosed) { + return; + } + peerZkUrl = url; + maybeScheduleRetry(); + } + ensureConnection(); + } + + // Schedule the periodic retry once, the first time the watcher has a peer to watch, so a watcher + // that is never configured starts no thread. Call under stateLock. + private void maybeScheduleRetry() { + if ( + retryScheduled || watcherClosed || retryIntervalSec <= 0 || StringUtils.isBlank(peerZkUrl) + ) { + return; + } + long initialDelaySec = ThreadLocalRandom.current().nextLong(1, retryIntervalSec + 1); + retryExecutor.scheduleAtFixedRate(this::retryIfBlind, initialDelaySec, retryIntervalSec, + TimeUnit.SECONDS); + retryScheduled = true; + } + + /** Reconcile off the caller's thread; used from the Curator event thread. */ + void reconfigureAsync(String url) { + ScheduledExecutorService ex = retryExecutor; + if (ex == null) { + reconfigure(url); + return; + } + try { + ex.execute(() -> reconfigure(url)); + } catch (RejectedExecutionException e) { + LOGGER.debug("Peer reconfigure skipped for HA group {}: watcher closing", haGroupName); + } + } + + /** Current peer record, or null when the peer is not visible. */ + HAGroupStoreRecord getCurrentPeerRecord() { + // Read the cache under the lock: close() nulls the field and closes the cache only after + // acquiring this lock, so the O(1) in-memory read here always sees a live cache. + synchronized (stateLock) { + return HAGroupStoreCacheUtil.recordAndStatAt(cache, toPath(haGroupName)).getLeft(); + } + } Review Comment: getCurrentPeerRecord() claims to return null when the peer is not visible, but it currently returns the cached record even when the watcher is BLIND/UNKNOWN. That can leak a stale peer role/state to callers (e.g., ClusterRoleRecord derivation) during peer-ZK outages and contradicts the method’s own contract. ########## 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 { + synchronized (localDegradedStandbyLock) { + HAGroupStoreRecord local = getHAGroupStoreRecord(); + if ( + localDegradedStandbyActive && local != null + && local.getHAGroupState() == HAGroupState.STANDBY + ) { + return local.withHAGroupState(HAGroupState.DEGRADED_STANDBY); + } + return local; } - return Pair.of(null, null); } - private Pair<HAGroupStoreRecord, Stat> - extractHAGroupStoreRecordOrNull(final ChildData childData) { - if (childData != null) { - byte[] data = childData.getData(); - return Pair.of(HAGroupStoreRecord.fromJson(data).orElse(null), childData.getStat()); + /** + * Receives peer observations from {@link PeerClusterWatcher}: peer record changes (already + * de-duplicated, with one forced redelivery on reconnect) and peer visible/blind transitions. + * Visibility transitions drive this RegionServer's local-only effective state; they are in-memory + * only and never written to the shared HA record. + */ + private final class PeerListener implements PeerClusterWatcher.PeerStateListener { + @Override + public void onPeerStateChanged(HAGroupStoreRecord peerRecord, Stat stat) { + HAGroupState fromState = lastKnownPeerState; + HAGroupState toState = peerRecord.getHAGroupState(); + lastKnownPeerState = toState; + LOGGER.info("Detected state transition for HA group {} from {} to {} on PEER cluster", + haGroupName, fromState, toState); + notifySubscribers(fromState, toState, stat != null ? stat.getMtime() : 0L, ClusterType.PEER, + peerRecord.getLastSyncStateTimeInMs()); + // A peer role change alters the combined legacy CRR; re-derive it off this callback thread. + offloadLegacyCrrSync(); + } + + @Override + public void onPeerVisible() { + clearLocalDegradedStandbyIfStillStandby(); + } + + @Override + public void onPeerBlind() { + presentLocalDegradedStandbyIfStandby(); + } + } + + /** Offload the legacy CRR sync (ZK + JDBC I/O) off the calling cache/event thread. */ + private void offloadLegacyCrrSync() { + ScheduledExecutorService syncExec = legacyCrrSyncExecutor; + if (syncExec != null) { + try { + syncExec.execute(this::syncLegacyCRRIfRoleChanged); + } catch (RejectedExecutionException ree) { + LOGGER.debug("Legacy CRR sync skipped for HA group {}: executor shut down", haGroupName); + } } - return Pair.of(null, null); } /** - * Closes the peer connection and cleans up peer-related resources. + * If this cluster is STANDBY and the peer is blind, present an in-memory DEGRADED_STANDBY (the + * shared HA record in ZK and the local PathChildrenCache are untouched) and return true; + * otherwise return false. Idempotent. Serialized with clear via localTransitionLock. */ - private void closePeerConnection() { - try { - if (peerPathChildrenCache != null) { - peerPathChildrenCache.close(); - peerPathChildrenCache = null; + private boolean presentLocalDegradedStandbyIfStandby() { + synchronized (localTransitionLock) { + HAGroupStoreRecord local; + synchronized (localDegradedStandbyLock) { + local = readLocalRecordQuietly(); + if ( + localDegradedStandbyActive || local == null + || local.getHAGroupState() != HAGroupState.STANDBY || !peerWatcher.isBlind() + ) { + return false; + } + localDegradedStandbyActive = true; } - if (peerPhoenixHaAdmin != null) { - peerPhoenixHaAdmin.close(); - peerPhoenixHaAdmin = null; + LOGGER.warn("Peer not visible for HA group {}; presenting local DEGRADED_STANDBY " + + "(reason=peer-blind)", haGroupName); + notifySubscribers(HAGroupState.STANDBY, HAGroupState.DEGRADED_STANDBY, + System.currentTimeMillis(), ClusterType.LOCAL, local.getLastSyncStateTimeInMs()); + return true; + } + } + + /** + * The peer is visible again: lift the local-only degrade. If this cluster is still STANDBY, + * notify subscribers that the effective local state is STANDBY again. If it failed over while + * blind, the real state already reached subscribers. + */ + private void clearLocalDegradedStandbyIfStillStandby() { + synchronized (localTransitionLock) { + HAGroupStoreRecord local; + boolean recover; + synchronized (localDegradedStandbyLock) { + if (!localDegradedStandbyActive) { + return; + } + local = readLocalRecordQuietly(); + localDegradedStandbyActive = false; + recover = local != null && local.getHAGroupState() == HAGroupState.STANDBY; + } + if (recover) { + LOGGER.warn("Peer visible again for HA group {}; clearing local DEGRADED_STANDBY " + + "(reason=peer-blind)", haGroupName); Review Comment: Log message contradicts itself: this branch runs when the peer is visible again, but the reason tag still says "peer-blind". This makes troubleshooting harder because the logs don’t distinguish degrade vs recover causes. -- 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]
