apurtell commented on code in PR #2547: URL: https://github.com/apache/phoenix/pull/2547#discussion_r3501420810
########## 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(); + } + } + + /** True when the peer is not currently visible (unknown or lost). */ + boolean isBlind() { + return visibility != Visibility.VISIBLE; + } + + /** Whether the peer cache is currently built and live. */ + @VisibleForTesting + boolean hasPeerCache() { + synchronized (stateLock) { + return cache != null; + } + } + + /** Whether the periodic retry has been armed; scheduled lazily on the first configured peer. */ + @VisibleForTesting + boolean isRetryScheduled() { + synchronized (stateLock) { + return retryScheduled; + } + } + + /** Blocking, event-free rebuild of the peer cache (mirrors {@code PathChildrenCache.rebuild}). */ + void rebuild() { + PathChildrenCache c; + synchronized (stateLock) { + c = cache; + } + if (c != null) { + try { + c.rebuild(); + } catch (Exception e) { + LOGGER.error("Peer cache rebuild failed for HA group {}", haGroupName, e); + } + } + } + + @Override + public void close() { + ScheduledExecutorService ex; + synchronized (stateLock) { + watcherClosed = true; + ex = retryExecutor; + retryExecutor = null; + } + if (ex != null) { + MoreExecutors.shutdownAndAwaitTermination(ex, 5, TimeUnit.SECONDS); + } + closeConnection(); + } + + private void retryIfBlind() { + boolean needsBuild; + String desiredUrl; + long attempt = 0L; + synchronized (stateLock) { + needsBuild = !watcherClosed && StringUtils.isNotBlank(peerZkUrl) && cache == null; Review Comment: Recovery depends on the `cache` field being null, which could be problematic. `CuratorFramework` is configured with `HighAvailabilityGroup.RETRY_POLICY`, an exponential backoff retry, 5 retries, 10s max sleep, roughly a minute. If a peer ZK outage exceeds that time the watcher could stay blind indefinitely with cache != null and thus no path to healing. I think generally Curator will manage to reconnect and the watcher will recover, but there could be prolonged disruptions. We may want to consider a watchdog that, when blind for longer than some threshold, responds by tearing down the peer admin/cache and building it again. ########## 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() { Review Comment: Returning a stale value can have downstream effects. In HAGroupStoreClient#getHAGroupStoreRecordFromPeer() → getClusterRoleRecord(), the state state is visible to external clients building/refreshing routing decisions. In syncZKToSystemTable() we can persist a stale peer role into SYSTEM.HA_GROUP. In syncLegacyCRRIfRoleChanged() we can publish a stale combined CRR . In setHAGroupStatusIfNeeded() we might update post transition state with the stale peer role. This introduces fail open cases. I think you just need to add ```java if (isBlind()) { return null; } ``` at the top of the method. ########## phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/HAGroupStoreClient.java: ########## @@ -122,23 +120,36 @@ public class HAGroupStoreClient implements Closeable { private final String haGroupName; // PathChildrenCache for current cluster and HAGroupName private PathChildrenCache pathChildrenCache = null; - // PathChildrenCache for peer cluster and HAGroupName - private PathChildrenCache peerPathChildrenCache = null; + // Watches the peer cluster's HA record; owns the peer cache/admin, retry, and visibility. + private final PeerClusterWatcher peerWatcher; + // True while this RegionServer presents a local-only DEGRADED_STANDBY. The shared HA record is + // not changed; this only affects the effective local view and de-dupes the synthetic + // degrade/recover notifications. + private volatile boolean localDegradedStandbyActive = false; // Whether the client is healthy private volatile boolean isHealthy = false; // Configuration private final Configuration conf; // ZK URL for the current cluster and HAGroupName private String zkUrl; - // Peer Custom Event Listener - private final PathChildrenCacheListener peerCustomPathChildrenCacheListener; // Wait time for sync mode private final long waitTimeForSyncModeInMs; // Rotation time for a log private final long rotationTimeMs; // State tracking for transition detection private volatile HAGroupState lastKnownLocalState; + // Last peer state delivered to subscribers; used only to populate the "from" state in PEER + // notifications. Written from the peer watcher's single delivery thread. private volatile HAGroupState lastKnownPeerState; + // Last peer ZK URL applied to the peer watcher; used to skip a reconcile when it is unchanged. + private volatile String lastConfiguredPeerZKUrl; + // Guards the local-only DEGRADED_STANDBY flag; short critical sections only, and the lock + // getEffectiveHAGroupStoreRecord() uses. Notifications run outside this lock. + private final Object localDegradedStandbyLock = new Object(); Review Comment: localDegradedStandbyLock... A stuck peer rebuild can stall peer state delivery on the local side. `getEffectiveHAGroupStoreRecord()` takes localDegradedStandbyLock and then `fetchLocalRecordAndPopulateZKIfNeeded` -> `rebuild()` -> `peerWatcher.rebuild()`, which does a blocking `PathChildrenCache.rebuild()` against the peer ZK. Concurrently meanwhile a peer event delivering `setBlind`/`setVisible` could call `presentLocalDegradedStandbyIfStandby` or `clearLocalDegradedStandbyIfStillStandby`, both of which need to take `localDegradedStandbyLock`. Consider these mitigations In `getEffectiveHAGroupStoreRecord`, snapshot the local record into a local first, then take `localDegradedStandbyLock` only for making the decision. OR In `fetchLocalRecordAndPopulateZKIfNeeded`, skip `peerWatcher.rebuild()` , because the local cache is the only thing that needs to be rebuilt. ########## 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 { + + /** Review Comment: This javadoc is subtly wrong. onPeerVisible and onPeerBlind are invoked under lock, but onPeerStateChanged is invoked from deliver(), which is not holding transitionLock, only stateLock briefly, and the listener fires outside it. ########## 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, Review Comment: Consider passing the prior effective state explicitly. ########## 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); + notifySubscribers(HAGroupState.DEGRADED_STANDBY, HAGroupState.STANDBY, Review Comment: Same concern as above. This hard codes a from state of DEGRADED_STANDBY regardless of what the listener actually observed. ########## 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: This seems like a good suggestion. ########## 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 is held. 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]
