This is an automated email from the ASF dual-hosted git repository.
czy006 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 7b19b7eaa [Subtask]: Optimizations and troubleshooting for the
Master-Slave mode. (#4174)
7b19b7eaa is described below
commit 7b19b7eaaa6f5ddd809d75886e15916b7ab57126
Author: can <[email protected]>
AuthorDate: Thu Apr 16 10:09:54 2026 +0800
[Subtask]: Optimizations and troubleshooting for the Master-Slave mode.
(#4174)
* [Subtask]: Fixed several bugs in master-slave mode. #4171
* [Subtask]: Fixed several bugs in master-slave mode. #4171
* [Subtask]: Optimized based on CR feedback. #4171
---------
Co-authored-by: wardli <[email protected]>
---
.../apache/amoro/server/AmoroManagementConf.java | 18 +-
.../apache/amoro/server/AmoroServiceContainer.java | 10 +-
.../org/apache/amoro/server/AmsAssignService.java | 218 +++++++++++++++++----
.../amoro/server/DefaultOptimizingService.java | 3 +-
.../ha/DataBaseHighAvailabilityContainer.java | 6 +-
.../server/ha/ZkHighAvailabilityContainer.java | 25 ++-
.../amoro/server/table/AbstractTableMetrics.java | 12 +-
.../amoro/server/table/DefaultTableService.java | 206 ++++++++++++-------
.../apache/amoro/server/TestAmsAssignService.java | 68 ++++++-
.../server/TestHighAvailabilityContainer.java | 20 +-
.../server/ha/TestZkHighAvailabilityContainer.java | 20 +-
.../common/AbstractOptimizerOperator.java | 143 ++++++++------
.../amoro/optimizer/common/AmsNodeManager.java | 53 +++--
.../amoro/optimizer/common/OptimizerExecutor.java | 84 +++++++-
.../amoro/optimizer/common/OptimizerToucher.java | 34 +++-
.../optimizer/common/ThriftAmsNodeManager.java | 15 +-
docs/configuration/ams-config.md | 2 +-
17 files changed, 699 insertions(+), 238 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
index 98331ad33..9ac6fb08e 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroManagementConf.java
@@ -107,15 +107,6 @@ public class AmoroManagementConf {
.defaultValue("")
.withDescription("Optional LDAP bind password used when querying
role-mapping groups.");
- /** Enable master & slave mode, which supports horizontal scaling of AMS. */
- public static final ConfigOption<Boolean> USE_MASTER_SLAVE_MODE =
- ConfigOptions.key("use-master-slave-mode")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "This setting controls whether to enable the AMS horizontal
scaling feature, "
- + "which is currently under development and testing.");
-
public static final ConfigOption<Duration>
CATALOG_META_CACHE_EXPIRATION_INTERVAL =
ConfigOptions.key("catalog-meta-cache.expiration-interval")
.durationType()
@@ -340,6 +331,15 @@ public class AmoroManagementConf {
.defaultValue(java.time.Duration.ofSeconds(30))
.withDescription("TTL of HA lease.");
+ /** Enable master & slave mode, which supports horizontal scaling of AMS. */
+ public static final ConfigOption<Boolean> HA_USE_MASTER_SLAVE_MODE =
+ ConfigOptions.key("ha.use-master-slave-mode")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "This setting controls whether to enable the AMS horizontal
scaling feature, "
+ + "which is currently under development and testing.");
+
public static final ConfigOption<Integer> HA_BUCKET_ID_TOTAL_COUNT =
ConfigOptions.key("ha.bucket-id.total-count")
.intType()
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
index 768dab89f..0be364523 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmoroServiceContainer.java
@@ -18,7 +18,7 @@
package org.apache.amoro.server;
-import static
org.apache.amoro.server.AmoroManagementConf.USE_MASTER_SLAVE_MODE;
+import static
org.apache.amoro.server.AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE;
import io.javalin.Javalin;
import io.javalin.http.Context;
@@ -373,7 +373,7 @@ public class AmoroServiceContainer {
private void initConfig() throws Exception {
LOG.info("initializing configurations...");
new ConfigurationHelper().init();
- IS_MASTER_SLAVE_MODE = serviceConfig.getBoolean(USE_MASTER_SLAVE_MODE);
+ IS_MASTER_SLAVE_MODE = serviceConfig.getBoolean(HA_USE_MASTER_SLAVE_MODE);
}
public Configurations getServiceConfig() {
@@ -667,6 +667,12 @@ public class AmoroServiceContainer {
containerProperties.putIfAbsent(
OptimizerProperties.AMS_OPTIMIZER_URI,
AmsUtil.getAMSThriftAddress(serviceConfig,
Constants.THRIFT_OPTIMIZING_SERVICE_NAME));
+ // When master-slave mode is enabled, automatically inject the flag
into container
+ // properties to ensure the -msm argument is correctly passed when
starting optimizers.
+ if
(serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE)) {
+ containerProperties.putIfAbsent(
+ OptimizerProperties.OPTIMIZER_MASTER_SLAVE_MODE_ENABLED,
"true");
+ }
// put addition system properties
container.setProperties(containerProperties);
containerList.add(container);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/AmsAssignService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/AmsAssignService.java
index 186ba1bcf..018be60a7 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/AmsAssignService.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/AmsAssignService.java
@@ -97,7 +97,7 @@ public class AmsAssignService {
* Start the assignment service. Only works in master-slave mode and when
current node is leader.
*/
public void start() {
- if (!serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE)) {
+ if
(!serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE)) {
LOG.info("Master-slave mode is not enabled, skip starting bucket
assignment service");
return;
}
@@ -151,7 +151,20 @@ public class AmsAssignService {
detectNodeChanges(aliveNodes, currentAssignments, aliveNodeMap,
normalized.assignedNodes);
if (!change.needReassign()) {
- refreshLastUpdateTime(aliveNodes);
+ // Even if no node changes, check if current assignments are balanced.
+ // This handles cases where previous rebalancing was incomplete or
+ // assignments became uneven due to historical reasons.
+ if (isUnbalanced(aliveNodes, normalized.assignments)) {
+ LOG.info(
+ "No node changes detected, but current assignments are
unbalanced. Performing rebalance...");
+ List<String> allBuckets = generateBucketIds();
+ Map<AmsServerInfo, List<String>> newAssignments =
+ buildNewAssignments(aliveNodes, new HashSet<>(),
normalized.assignments);
+ rebalanceExistingAssignments(aliveNodes, allBuckets, newAssignments);
+ persistAssignments(newAssignments);
+ } else {
+ refreshLastUpdateTime(aliveNodes);
+ }
return;
}
@@ -175,8 +188,92 @@ public class AmsAssignService {
}
/**
- * Redistribute buckets incrementally to alive nodes using round-robin. This
minimizes bucket
- * migration by only redistributing buckets from offline nodes.
+ * Check if the current assignments are unbalanced. Assignments are
considered unbalanced if the
+ * difference between the max and min bucket count across nodes exceeds 1.
+ *
+ * @param aliveNodes List of alive nodes
+ * @param normalizedAssignments Current normalized assignments
+ * @return true if assignments are unbalanced
+ */
+ private boolean isUnbalanced(
+ List<AmsServerInfo> aliveNodes, Map<AmsServerInfo, List<String>>
normalizedAssignments) {
+ if (aliveNodes.size() <= 1) {
+ return false;
+ }
+ int maxCount = Integer.MIN_VALUE;
+ int minCount = Integer.MAX_VALUE;
+ for (AmsServerInfo node : aliveNodes) {
+ int count = normalizedAssignments.getOrDefault(node, new
ArrayList<>()).size();
+ maxCount = Math.max(maxCount, count);
+ minCount = Math.min(minCount, count);
+ }
+ // Balanced means the difference between max and min should be at most 1
+ return (maxCount - minCount) > 1;
+ }
+
+ /**
+ * Rebalance existing assignments to ensure even distribution across all
alive nodes. Moves
+ * buckets from nodes with too many to nodes with too few, minimizing total
migration.
+ *
+ * @param aliveNodes All alive nodes
+ * @param allBuckets All bucket IDs
+ * @param currentAssignments Current assignments map (will be modified)
+ */
+ private void rebalanceExistingAssignments(
+ List<AmsServerInfo> aliveNodes,
+ List<String> allBuckets,
+ Map<AmsServerInfo, List<String>> currentAssignments) {
+ int totalBuckets = allBuckets.size();
+ int totalNodes = aliveNodes.size();
+ int targetPerNode = totalBuckets / totalNodes;
+ int remainder = totalBuckets % totalNodes;
+
+ // Collect excess buckets from nodes that have too many
+ List<String> excessBuckets = new ArrayList<>();
+
+ // Sort nodes by current bucket count descending, so we take from the most
loaded first
+ List<AmsServerInfo> sortedNodes = new ArrayList<>(aliveNodes);
+ sortedNodes.sort(
+ (n1, n2) -> {
+ int c1 = currentAssignments.getOrDefault(n1, new
ArrayList<>()).size();
+ int c2 = currentAssignments.getOrDefault(n2, new
ArrayList<>()).size();
+ return Integer.compare(c2, c1);
+ });
+
+ // First pass: determine each node's target and collect excess buckets
+ // The first 'remainder' nodes (sorted by count desc) get targetPerNode +
1,
+ // the rest get targetPerNode
+ int remainderSlots = remainder;
+ for (AmsServerInfo node : sortedNodes) {
+ int nodeTarget = targetPerNode + (remainderSlots > 0 ? 1 : 0);
+ List<String> buckets = currentAssignments.getOrDefault(node, new
ArrayList<>());
+ if (buckets.size() > nodeTarget) {
+ // Remove excess buckets from the end of the list
+ int excess = buckets.size() - nodeTarget;
+ for (int i = 0; i < excess; i++) {
+ excessBuckets.add(buckets.remove(buckets.size() - 1));
+ }
+ LOG.info(
+ "Collected {} excess buckets from node {} (had: {}, target: {})",
+ excess,
+ node,
+ buckets.size() + excess,
+ nodeTarget);
+ }
+ if (remainderSlots > 0) {
+ remainderSlots--;
+ }
+ }
+
+ // Second pass: distribute excess buckets to nodes that have fewer than
target
+ if (!excessBuckets.isEmpty()) {
+ redistributeBucketsIncrementally(aliveNodes, excessBuckets,
currentAssignments);
+ }
+ }
+
+ /**
+ * Redistribute buckets incrementally to alive nodes, always assigning each
bucket to the node
+ * with the fewest buckets. This ensures even distribution regardless of the
current state.
*
* @param aliveNodes List of alive nodes
* @param bucketsToRedistribute Buckets to redistribute (from offline nodes)
@@ -190,12 +287,18 @@ public class AmsAssignService {
return;
}
- // Distribute buckets using round-robin to minimize migration
- int nodeIndex = 0;
+ // Assign each bucket to the node with the fewest buckets to ensure even
distribution
for (String bucketId : bucketsToRedistribute) {
- AmsServerInfo node = aliveNodes.get(nodeIndex % aliveNodes.size());
- currentAssignments.get(node).add(bucketId);
- nodeIndex++;
+ AmsServerInfo minNode = null;
+ int minCount = Integer.MAX_VALUE;
+ for (AmsServerInfo node : aliveNodes) {
+ int count = currentAssignments.get(node).size();
+ if (count < minCount) {
+ minCount = count;
+ minNode = node;
+ }
+ }
+ currentAssignments.get(minNode).add(bucketId);
}
}
@@ -219,13 +322,24 @@ public class AmsAssignService {
return;
}
- // Calculate how many buckets each new node should get
- int bucketsPerNewNode = targetBucketsPerNode;
+ // Determine which alive nodes already have extra buckets (i.e.
targetBucketsPerNode + 1)
+ // so that remainder slots are distributed fairly across all nodes, not
just new ones.
+ int remainderSlotsTaken = 0;
+ for (AmsServerInfo node : aliveNodes) {
+ if (!newNodes.contains(node)) {
+ int count = currentAssignments.getOrDefault(node, new
ArrayList<>()).size();
+ if (count > targetBucketsPerNode) {
+ remainderSlotsTaken++;
+ }
+ }
+ }
+
int newNodeIndex = 0;
for (AmsServerInfo newNode : newNodes) {
- // First 'remainder' nodes get one extra bucket
- int targetForNewNode = bucketsPerNewNode + (newNodeIndex < remainder ? 1
: 0);
- int currentCount = currentAssignments.get(newNode).size();
+ // Assign extra bucket only if there are remaining remainder slots
+ int targetForNewNode =
+ targetBucketsPerNode + ((remainderSlotsTaken + newNodeIndex) <
remainder ? 1 : 0);
+ int currentCount = currentAssignments.getOrDefault(newNode, new
ArrayList<>()).size();
int needed = targetForNewNode - currentCount;
if (needed > 0) {
@@ -380,41 +494,61 @@ public class AmsAssignService {
Map<AmsServerInfo, List<String>> currentAssignments,
Map<String, AmsServerInfo> aliveNodeMap,
Set<AmsServerInfo> currentAssignedNodes) {
- Set<AmsServerInfo> aliveNodeSet = new HashSet<>(aliveNodes);
- Set<AmsServerInfo> newNodes = new HashSet<>(aliveNodeSet);
- newNodes.removeAll(currentAssignedNodes);
-
- Set<AmsServerInfo> offlineNodes = new HashSet<>();
- for (AmsServerInfo storedNode : currentAssignments.keySet()) {
- if (!aliveNodeMap.containsKey(getNodeKey(storedNode))) {
- offlineNodes.add(storedNode);
+ // Build nodeKey sets for comparison instead of relying on
AmsServerInfo.equals(),
+ // because ZkBucketAssignStore.parseNodeKey() only sets host and
thriftBindPort (restBindPort
+ // is null), which would cause equals() mismatch with the full
AmsServerInfo from
+ // getAliveNodes().
+ Set<String> assignedNodeKeys = new HashSet<>();
+ for (AmsServerInfo node : currentAssignedNodes) {
+ assignedNodeKeys.add(getNodeKey(node));
+ }
+ Set<AmsServerInfo> newNodes = new HashSet<>();
+ for (AmsServerInfo aliveNode : aliveNodes) {
+ if (!assignedNodeKeys.contains(getNodeKey(aliveNode))) {
+ newNodes.add(aliveNode);
}
}
+ Set<AmsServerInfo> offlineNodes = new HashSet<>();
long currentTime = System.currentTimeMillis();
Set<String> aliveNodeKeys = new HashSet<>();
for (AmsServerInfo node : aliveNodes) {
aliveNodeKeys.add(getNodeKey(node));
}
+
for (AmsServerInfo node : currentAssignedNodes) {
String nodeKey = getNodeKey(node);
+ // If the node is currently alive, it should never be considered offline,
+ // even if its lastUpdateTime is stale (e.g. node restarted within
nodeOfflineTimeoutMs
+ // but lastUpdateTime was not refreshed while it was down).
if (aliveNodeKeys.contains(nodeKey)) {
- try {
- long lastUpdateTime = assignStore.getLastUpdateTime(node);
- if (lastUpdateTime > 0 && (currentTime - lastUpdateTime) >
nodeOfflineTimeoutMs) {
- for (AmsServerInfo storedNode : currentAssignments.keySet()) {
- if (getNodeKey(storedNode).equals(nodeKey)) {
- offlineNodes.add(storedNode);
- break;
- }
- }
- LOG.warn(
- "Node {} is considered offline due to timeout. Last update:
{}",
- node,
- lastUpdateTime);
- }
- } catch (BucketAssignStoreException e) {
- LOG.warn("Failed to get last update time for node {}, treating as
offline", node, e);
+ continue;
+ }
+ try {
+ long lastUpdateTime = assignStore.getLastUpdateTime(node);
+ boolean shouldMarkOffline;
+ if (lastUpdateTime <= 0) {
+ // Missing timestamp means the node's saveAssignments never
completed its
+ // updateLastUpdateTime call (e.g. leader crashed between the two ZK
writes)
+ // or the store data was corrupted. Since the node is already
absent from
+ // the alive list, treat it as offline immediately to avoid stranded
buckets.
+ shouldMarkOffline = true;
+ LOG.warn(
+ "Node {} is considered offline (missing last update time, not in
alive list)", node);
+ } else if ((currentTime - lastUpdateTime) > nodeOfflineTimeoutMs) {
+ shouldMarkOffline = true;
+ LOG.warn(
+ "Node {} is considered offline due to timeout. Last update: {}",
+ node,
+ lastUpdateTime);
+ } else {
+ shouldMarkOffline = false;
+ LOG.debug(
+ "Node {} is not in alive list but heartbeat not timeout (last
update: {}), waiting for timeout",
+ node,
+ lastUpdateTime);
+ }
+ if (shouldMarkOffline) {
for (AmsServerInfo storedNode : currentAssignments.keySet()) {
if (getNodeKey(storedNode).equals(nodeKey)) {
offlineNodes.add(storedNode);
@@ -422,6 +556,14 @@ public class AmsAssignService {
}
}
}
+ } catch (BucketAssignStoreException e) {
+ LOG.warn("Failed to get last update time for node {}, treating as
offline", node, e);
+ for (AmsServerInfo storedNode : currentAssignments.keySet()) {
+ if (getNodeKey(storedNode).equals(nodeKey)) {
+ offlineNodes.add(storedNode);
+ break;
+ }
+ }
}
}
return new NodeChangeResult(newNodes, offlineNodes);
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
index e17672768..efc1c375b 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/DefaultOptimizingService.java
@@ -157,7 +157,8 @@ public class DefaultOptimizingService extends
StatedPersistentBase
this.bucketAssignStore = bucketAssignStore;
this.haContainer = haContainer;
this.isMasterSlaveMode =
- haContainer != null &&
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+ haContainer != null
+ &&
serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE);
this.tableHandlerChain = new TableRuntimeHandlerImpl();
this.planExecutor =
Executors.newCachedThreadPool(
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/ha/DataBaseHighAvailabilityContainer.java
b/amoro-ams/src/main/java/org/apache/amoro/server/ha/DataBaseHighAvailabilityContainer.java
index 7d6b6b37e..1aeaba583 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/ha/DataBaseHighAvailabilityContainer.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/ha/DataBaseHighAvailabilityContainer.java
@@ -141,7 +141,8 @@ public class DataBaseHighAvailabilityContainer extends
PersistentBase
@Override
public void registerAndElect() throws Exception {
- boolean isMasterSlaveMode =
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+ boolean isMasterSlaveMode =
+ serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE);
if (!isMasterSlaveMode) {
LOG.debug("Master-slave mode is not enabled, skip node registration");
return;
@@ -220,7 +221,8 @@ public class DataBaseHighAvailabilityContainer extends
PersistentBase
}
// Remove this node from bucket_assignments so the leader immediately
stops seeing it
// as alive without waiting for the heartbeat TTL to expire.
- boolean isMasterSlaveMode =
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+ boolean isMasterSlaveMode =
+ serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE);
if (isMasterSlaveMode) {
try {
String nodeKey = getNodeKey();
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/ha/ZkHighAvailabilityContainer.java
b/amoro-ams/src/main/java/org/apache/amoro/server/ha/ZkHighAvailabilityContainer.java
index 5c93a6777..9d557737c 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/ha/ZkHighAvailabilityContainer.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/ha/ZkHighAvailabilityContainer.java
@@ -67,7 +67,7 @@ public class ZkHighAvailabilityContainer implements
HighAvailabilityContainer, L
private String registeredNodePath;
public ZkHighAvailabilityContainer(Configurations serviceConfig) throws
Exception {
- this.isMasterSlaveMode =
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+ this.isMasterSlaveMode =
serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE);
if (serviceConfig.getBoolean(AmoroManagementConf.HA_ENABLE)) {
String zkServerAddress =
serviceConfig.getString(AmoroManagementConf.HA_ZOOKEEPER_ADDRESS);
int zkSessionTimeout =
@@ -212,6 +212,29 @@ public class ZkHighAvailabilityContainer implements
HighAvailabilityContainer, L
tableServiceServerInfo.toString(),
optimizingServiceServerInfo.toString());
followerLatch = new CountDownLatch(1);
+ // Write server info to ZK master paths on leadership gained, so that both
master-slave mode
+ // (which skips waitLeaderShip()) and active-standby mode have the correct
JSON data in ZK.
+ try {
+ CuratorOp tableServiceMasterPathOp =
+ zkClient
+ .transactionOp()
+ .setData()
+ .forPath(
+ tableServiceMasterPath,
+ JacksonUtil.toJSONString(tableServiceServerInfo)
+ .getBytes(StandardCharsets.UTF_8));
+ CuratorOp optimizingServiceMasterPathOp =
+ zkClient
+ .transactionOp()
+ .setData()
+ .forPath(
+ optimizingServiceMasterPath,
+ JacksonUtil.toJSONString(optimizingServiceServerInfo)
+ .getBytes(StandardCharsets.UTF_8));
+ zkClient.transaction().forOperations(tableServiceMasterPathOp,
optimizingServiceMasterPathOp);
+ } catch (Exception e) {
+ LOG.error("Failed to write server info to ZK master paths on leadership
gained", e);
+ }
}
@Override
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
index 74e52f0b7..4778a2044 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/AbstractTableMetrics.java
@@ -54,8 +54,16 @@ public abstract class AbstractTableMetrics {
public void register(MetricRegistry registry) {
if (globalRegistry == null) {
- registerMetrics(registry);
- globalRegistry = registry;
+ try {
+ registerMetrics(registry);
+ globalRegistry = registry;
+ } catch (Exception e) {
+ // Roll back any metrics that were partially registered before the
failure so that the
+ // next retry call finds a clean state (globalRegistry stays null,
partial keys removed).
+ registeredMetricKeys.forEach(registry::unregister);
+ registeredMetricKeys.clear();
+ throw e;
+ }
}
}
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
index c64b7d923..6e916bfd7 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableService.java
@@ -105,6 +105,27 @@ public class DefaultTableService extends PersistentBase
implements TableService
// Master-slave mode related fields
private ScheduledExecutorService bucketTableSyncScheduler;
private volatile List<String> assignedBucketIds = new ArrayList<>();
+
+ /**
+ * Serializes the "read bucket counts → select bucket" critical section so
that concurrent
+ * table-explorer threads cannot pick the same bucket ID within the same
exploration cycle.
+ *
+ * <p>Using only a DB-side count is insufficient: {@code insertRuntime} may
be executed inside a
+ * larger uncommitted {@code doAsTransaction}, so a concurrent thread
reading the DB before that
+ * transaction commits still sees the old count and picks the same bucket.
The {@code
+ * pendingBucketAssignments} map below compensates by tracking in-memory
reservations that have
+ * been decided but may not yet be visible in the DB.
+ */
+ private final Object bucketAssignLock = new Object();
+
+ /**
+ * Tracks bucket-ID reservations made during the current exploration cycle
that have been decided
+ * (and recorded in-memory) but whose DB writes may not yet be committed.
This map is always
+ * accessed under {@code bucketAssignLock} and is cleared at the start of
each exploration cycle
+ * so that it never drifts from the committed DB state across cycles.
+ */
+ private final Map<String, Integer> pendingBucketAssignments = new
HashMap<>();
+
private final long bucketTableSyncInterval;
public DefaultTableService(
@@ -128,7 +149,7 @@ public class DefaultTableService extends PersistentBase
implements TableService
this.tableRuntimeFactory.withTableLoader(this::loadTable);
this.haContainer = haContainer;
this.bucketAssignStore = bucketAssignStore;
- this.isMasterSlaveMode =
configuration.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE);
+ this.isMasterSlaveMode =
configuration.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE);
this.bucketTableSyncInterval =
configuration.get(AmoroManagementConf.HA_BUCKET_TABLE_SYNC_INTERVAL).toMillis();
}
@@ -240,6 +261,9 @@ public class DefaultTableService extends PersistentBase
implements TableService
continue;
}
List<TableRuntimeState> states =
statesMap.get(tableRuntimeMeta.getTableId());
+ if (states == null) {
+ states = Collections.emptyList();
+ }
Optional<TableRuntime> tableRuntime =
createTableRuntime(identifier, tableRuntimeMeta, states);
if (!tableRuntime.isPresent()) {
@@ -404,7 +428,8 @@ public class DefaultTableService extends PersistentBase
implements TableService
return a;
}));
- // Find tables that should be added (in DB but not in memory)
+ // Snapshot the current in-memory table IDs for the removal-detection
loop below.
+ // The "add" loop uses a live containsKey check so it does not rely on
this snapshot.
Set<Long> currentTableIds = new HashSet<>(tableRuntimeMap.keySet());
Set<Long> dbTableIds =
tableRuntimeMetaList.stream()
@@ -414,31 +439,34 @@ public class DefaultTableService extends PersistentBase
implements TableService
// Add new tables
for (TableRuntimeMeta tableRuntimeMeta : tableRuntimeMetaList) {
Long tableId = tableRuntimeMeta.getTableId();
- if (!currentTableIds.contains(tableId)) {
- ServerTableIdentifier identifier = identifierMap.get(tableId);
- if (identifier == null) {
- LOG.warn("No available table identifier found for table runtime
meta id={}", tableId);
- continue;
- }
- List<TableRuntimeState> states = statesMap.get(tableId);
- // Use empty list if states is null to avoid NullPointerException
- if (states == null) {
- states = Collections.emptyList();
- }
- Optional<TableRuntime> tableRuntime =
- createTableRuntime(identifier, tableRuntimeMeta, states);
- if (tableRuntime.isPresent()) {
- TableRuntime runtime = tableRuntime.get();
-
runtime.registerMetric(MetricManager.getInstance().getGlobalRegistry());
- tableRuntimeMap.put(tableId, runtime);
- if (headHandler != null) {
- AmoroTable<?> table = loadTable(identifier);
- if (table != null) {
- headHandler.fireTableAdded(table, runtime);
- }
+ // Use a live containsKey check instead of the snapshot to guard
against any concurrent
+ // addition (e.g. from exploreTableRuntimes running on a parallel
thread).
+ if (tableRuntimeMap.containsKey(tableId)) {
+ continue;
+ }
+ ServerTableIdentifier identifier = identifierMap.get(tableId);
+ if (identifier == null) {
+ LOG.warn("No available table identifier found for table runtime meta
id={}", tableId);
+ continue;
+ }
+ List<TableRuntimeState> states = statesMap.get(tableId);
+ // Use empty list if states is null to avoid NullPointerException
+ if (states == null) {
+ states = Collections.emptyList();
+ }
+ Optional<TableRuntime> tableRuntime =
+ createTableRuntime(identifier, tableRuntimeMeta, states);
+ if (tableRuntime.isPresent()) {
+ TableRuntime runtime = tableRuntime.get();
+
runtime.registerMetric(MetricManager.getInstance().getGlobalRegistry());
+ tableRuntimeMap.put(tableId, runtime);
+ if (headHandler != null) {
+ AmoroTable<?> table = loadTable(identifier);
+ if (table != null) {
+ headHandler.fireTableAdded(table, runtime);
}
- LOG.info("Added table {} for bucketId {}", tableId,
tableRuntimeMeta.getBucketId());
}
+ LOG.info("Added table {} for bucketId {}", tableId,
tableRuntimeMeta.getBucketId());
}
}
@@ -471,11 +499,17 @@ public class DefaultTableService extends PersistentBase
implements TableService
if (headHandler != null) {
headHandler.fireTableRemoved(tableRuntime);
}
- tableRuntime.dispose();
+ // Only unload from memory; do NOT call dispose() which would
delete the table from DB.
+ // This node is simply no longer responsible for the table (bucket
reassigned or the
+ // runtime row was already cleaned up elsewhere). Deleting from DB
here would destroy
+ // a live table managed by another node.
+ tableRuntime.unregisterMetric();
tableRuntimeMap.remove(tableId);
- LOG.info("Removed table {} as it's no longer assigned to this
node", tableId);
+ LOG.info(
+ "Unloaded table {} runtime from this node (bucket no longer
assigned here)",
+ tableId);
} catch (Exception e) {
- LOG.error("Error occurred while removing tableRuntime of table
{}", tableId, e);
+ LOG.error("Error occurred while unloading tableRuntime of table
{}", tableId, e);
}
}
}
@@ -485,10 +519,12 @@ public class DefaultTableService extends PersistentBase
implements TableService
}
/**
- * Assign bucketId to a table using min-heap strategy. This ensures tables
are evenly distributed
- * across bucketIds.
+ * Selects the bucket with the fewest assigned tables using a min-heap.
*
- * @return The assigned bucketId, or null if assignment fails
+ * <p><b>Must be called while holding {@code bucketAssignLock}.</b> The
method merges committed DB
+ * counts with {@code pendingBucketAssignments} (reservations made in the
current exploration
+ * cycle whose DB writes may not yet be committed) so that concurrent
callers within the same
+ * cycle always see each other's reservations and never select the same
bucket.
*/
private String assignBucketIdForTable() {
try {
@@ -497,9 +533,9 @@ public class DefaultTableService extends PersistentBase
implements TableService
TableRuntimeMapper.class,
(TableRuntimeMapper mapper) -> mapper.countTablesByBucketId());
- Map<String, Integer> bucketTableCount = new HashMap<>();
int bucketIdTotalCount =
serverConfiguration.getInteger(AmoroManagementConf.HA_BUCKET_ID_TOTAL_COUNT);
+ Map<String, Integer> bucketTableCount = new HashMap<>();
for (int i = 1; i <= bucketIdTotalCount; i++) {
bucketTableCount.put(String.valueOf(i), 0);
}
@@ -511,10 +547,15 @@ public class DefaultTableService extends PersistentBase
implements TableService
}
}
+ // Overlay in-memory reservations from this exploration cycle: these
writes may not yet be
+ // committed to the DB, so the DB count alone would be stale for
concurrent callers.
+ pendingBucketAssignments.forEach(
+ (bucket, pending) -> bucketTableCount.merge(bucket, pending,
Integer::sum));
+
PriorityQueue<Map.Entry<String, Integer>> minHeap =
new PriorityQueue<>(
Comparator.<Map.Entry<String,
Integer>>comparingInt(Map.Entry::getValue)
- .thenComparing(Map.Entry::getKey));
+ .thenComparingInt(e -> Integer.parseInt(e.getKey())));
for (Map.Entry<String, Integer> entry : bucketTableCount.entrySet()) {
minHeap.offer(new AbstractMap.SimpleEntry<>(entry.getKey(),
entry.getValue()));
@@ -523,11 +564,12 @@ public class DefaultTableService extends PersistentBase
implements TableService
if (!minHeap.isEmpty()) {
Map.Entry<String, Integer> selected = minHeap.poll();
String assignedBucketId = selected.getKey();
- int tableCount = selected.getValue();
+ // Record reservation immediately so the next caller (within the same
lock-cycle) sees it.
+ pendingBucketAssignments.merge(assignedBucketId, 1, Integer::sum);
LOG.debug(
- "Assigned bucketId {} to new table (min table count: {})",
+ "Assigned bucketId {} to new table (effective count: {})",
assignedBucketId,
- tableCount);
+ selected.getValue());
return assignedBucketId;
}
} catch (Exception e) {
@@ -550,6 +592,10 @@ public class DefaultTableService extends PersistentBase
implements TableService
throw new IllegalStateException("TableService is not initialized");
}
long start = System.currentTimeMillis();
+ // Reset the in-memory pending map so the new cycle starts from committed
DB state.
+ synchronized (bucketAssignLock) {
+ pendingBucketAssignments.clear();
+ }
List<ServerCatalog> externalCatalogs = catalogManager.getServerCatalogs();
List<String> externalCatalogNames =
externalCatalogs.stream().map(ServerCatalog::name).collect(Collectors.toList());
@@ -700,6 +746,23 @@ public class DefaultTableService extends PersistentBase
implements TableService
}
});
taskFutures.forEach(CompletableFuture::join);
+
+ // In master-slave mode, historical tables already registered in Amoro DB
may have a null
+ // bucket_id if they predate the master-slave feature. Unlike new tables
(handled above via
+ // syncTable), these tables are already in serverTableIdentifiers and are
skipped by the
+ // Sets.difference logic. We must explicitly iterate over them here so the
leader can assign
+ // bucket IDs, mirroring the behaviour of exploreInternalCatalog.
+ serverTableIdentifiers.values().stream()
+ .filter(i -> !tableRuntimeMap.containsKey(i.getId()))
+ .filter(i -> tableIdentifiers.contains(new TableIdentity(i)))
+ .forEach(
+ i -> {
+ try {
+ triggerTableAdded(externalCatalog, i);
+ } catch (Exception e) {
+ LOG.error("TableExplorer failed to assign bucketId for
existing table {}", i, e);
+ }
+ });
}
private void exploreInternalCatalog(InternalCatalog internalCatalog) {
@@ -802,7 +865,6 @@ public class DefaultTableService extends PersistentBase
implements TableService
// In master-slave mode (fully wired), assign bucketId to the table if
it's not assigned yet.
// Only leader node should assign bucketIds; follower may still persist
the table with null
// bucketId (e.g. onTableCreated on follower), and leader will assign
later via exploration.
- String assignedBucketId = null;
if (isMasterSlaveMode && haContainer != null && bucketAssignStore != null)
{
if (haContainer.hasLeadership()) {
TableRuntimeMeta existingMeta =
@@ -814,40 +876,51 @@ public class DefaultTableService extends PersistentBase
implements TableService
if (existingMeta.getBucketId() != null) {
return true; // already assigned
}
- assignedBucketId = assignBucketIdForTable();
- if (assignedBucketId != null) {
- String finalAssignedBucketId = assignedBucketId;
- long updated =
- updateAs(
- TableRuntimeMapper.class,
- mapper ->
- mapper.updateBucketIdIfNull(
- serverTableIdentifier.getId(),
finalAssignedBucketId));
- if (updated == 1) {
- LOG.info(
- "Assigned bucketId {} to existing table {} (was null)",
- assignedBucketId,
- serverTableIdentifier);
- } else {
- LOG.debug(
- "Skipped backfill bucketId {} for table {} (row missing or
bucket_id already set)",
- assignedBucketId,
- serverTableIdentifier);
+ // Existing row with null bucket_id: assign and update atomically so
that concurrent
+ // table-explorer threads reading identical DB counts cannot select
the same bucket.
+ synchronized (bucketAssignLock) {
+ String assignedBucketId = assignBucketIdForTable();
+ if (assignedBucketId != null) {
+ long updated =
+ updateAs(
+ TableRuntimeMapper.class,
+ mapper ->
+ mapper.updateBucketIdIfNull(
+ serverTableIdentifier.getId(),
assignedBucketId));
+ if (updated == 1) {
+ LOG.info(
+ "Assigned bucketId {} to existing table {} (was null)",
+ assignedBucketId,
+ serverTableIdentifier);
+ } else {
+ LOG.debug(
+ "Skipped backfill bucketId {} for table {} (row missing or
bucket_id already set)",
+ assignedBucketId,
+ serverTableIdentifier);
+ }
}
}
return true; // handled existing row (assigned or failed to assign)
}
- assignedBucketId = assignBucketIdForTable();
- if (assignedBucketId != null) {
- meta.setBucketId(assignedBucketId);
- LOG.info("Assigned bucketId {} to table {}", assignedBucketId,
serverTableIdentifier);
- } else {
- LOG.warn(
- "Failed to assign bucketId to table {}, will be assigned later",
- serverTableIdentifier);
+ // New table: assign bucket and insert runtime atomically so that
concurrent threads
+ // cannot read identical DB counts and select the same bucket.
+ synchronized (bucketAssignLock) {
+ String assignedBucketId = assignBucketIdForTable();
+ if (assignedBucketId != null) {
+ meta.setBucketId(assignedBucketId);
+ LOG.info("Assigned bucketId {} to table {}", assignedBucketId,
serverTableIdentifier);
+ } else {
+ LOG.warn(
+ "Failed to assign bucketId to table {}, will be assigned
later",
+ serverTableIdentifier);
+ }
+ doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta));
}
+ return true;
}
- // else: follower does not assign; meta.bucketId remains null until
leader assigns
+ // Follower: persist the table with null bucketId; leader will assign
later.
+ doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta));
+ return true;
}
doAs(TableRuntimeMapper.class, mapper -> mapper.insertRuntime(meta));
@@ -855,9 +928,6 @@ public class DefaultTableService extends PersistentBase
implements TableService
// Only skip local runtime creation when master-slave mode is fully wired
(bucketAssignStore
// is non-null). When bucketAssignStore is null (e.g. 3-arg test
constructor), fall through
// and create the runtime in memory as in non-master-slave mode.
- if (isMasterSlaveMode && haContainer != null && bucketAssignStore != null)
{
- return true;
- }
Optional<TableRuntime> tableRuntimeOpt =
createTableRuntime(serverTableIdentifier, meta,
Collections.emptyList());
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsAssignService.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsAssignService.java
index e441a6d1f..91fad6436 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsAssignService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestAmsAssignService.java
@@ -78,9 +78,9 @@ public class TestAmsAssignService {
serviceConfig.setBoolean(AmoroManagementConf.HA_ENABLE, true);
serviceConfig.setString(AmoroManagementConf.HA_ZOOKEEPER_ADDRESS,
"127.0.0.1:2181");
serviceConfig.setString(AmoroManagementConf.HA_CLUSTER_NAME,
"test-cluster");
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
serviceConfig.setInteger(AmoroManagementConf.HA_BUCKET_ID_TOTAL_COUNT,
100);
- serviceConfig.set(AmoroManagementConf.HA_NODE_OFFLINE_TIMEOUT,
java.time.Duration.ofMinutes(5));
+ serviceConfig.set(AmoroManagementConf.HA_NODE_OFFLINE_TIMEOUT,
java.time.Duration.ofMillis(50));
haContainer = createContainerWithMockZk();
@@ -381,6 +381,51 @@ public class TestAmsAssignService {
}
}
+ @Test
+ public void testOfflineNodeWithMissingLastUpdateTime() throws Exception {
+ // Verify that a node absent from the alive list with lastUpdateTime == 0
+ // is still reclaimed instead of being stranded forever.
+ haContainer.registerAndElect();
+ Configurations config2 = createNodeConfig("127.0.0.2", 1262, 1632);
+ HighAvailabilityContainer haContainer2 =
createContainerWithMockZk(config2);
+ haContainer2.registerAndElect();
+
+ try {
+ Thread.sleep(100);
+
+ // Initial assignment — both nodes get buckets
+ assignService.doAssign();
+ Map<AmsServerInfo, List<String>> initialAssignments =
mockAssignStore.getAllAssignments();
+ Assert.assertEquals("Should have 2 nodes", 2, initialAssignments.size());
+
+ // Simulate node2 going offline (remove from ZK)
+ mockZkState.deleteNodeByHost("127.0.0.2");
+
+ // Clear node2's lastUpdateTime to simulate the edge case where
+ // saveAssignments wrote the assignment but crashed before
+ // updateLastUpdateTime completed (two non-atomic ZK writes).
+ mockAssignStore.clearLastUpdateTime("127.0.0.2");
+
+ Thread.sleep(100);
+
+ // Trigger reassignment — node2 should be reclaimed even without a
timestamp
+ assignService.doAssign();
+
+ Map<AmsServerInfo, List<String>> newAssignments =
mockAssignStore.getAllAssignments();
+ Assert.assertEquals(
+ "Node with missing lastUpdateTime should be reclaimed", 1,
newAssignments.size());
+
+ List<String> remainingBuckets =
newAssignments.values().iterator().next();
+ Assert.assertEquals("All buckets should be redistributed", 100,
remainingBuckets.size());
+ } finally {
+ try {
+ haContainer2.close();
+ } catch (Exception e) {
+ // ignore
+ }
+ }
+ }
+
@Test
public void testServiceStartStop() {
// Test that service can start and stop without errors
@@ -428,9 +473,9 @@ public class TestAmsAssignService {
config.setBoolean(AmoroManagementConf.HA_ENABLE, true);
config.setString(AmoroManagementConf.HA_ZOOKEEPER_ADDRESS,
"127.0.0.1:2181");
config.setString(AmoroManagementConf.HA_CLUSTER_NAME, "test-cluster");
- config.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ config.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE, true);
config.setInteger(AmoroManagementConf.HA_BUCKET_ID_TOTAL_COUNT, 100);
- config.set(AmoroManagementConf.HA_NODE_OFFLINE_TIMEOUT,
java.time.Duration.ofMinutes(5));
+ config.set(AmoroManagementConf.HA_NODE_OFFLINE_TIMEOUT,
java.time.Duration.ofMillis(50));
return config;
}
@@ -490,7 +535,7 @@ public class TestAmsAssignService {
ZkHighAvailabilityContainer.class.getDeclaredField("isMasterSlaveMode");
isMasterSlaveModeField.setAccessible(true);
isMasterSlaveModeField.set(
- container,
config.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE));
+ container,
config.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE));
if (config.getBoolean(AmoroManagementConf.HA_ENABLE)) {
String haClusterName =
config.getString(AmoroManagementConf.HA_CLUSTER_NAME);
@@ -881,6 +926,19 @@ public class TestAmsAssignService {
return nodes;
}
+ /**
+ * Clear lastUpdateTime for all nodes matching the given host, simulating
a missing timestamp.
+ */
+ void clearLastUpdateTime(String host) {
+ List<String> toClear = new ArrayList<>();
+ for (String nodeKey : lastUpdateTimes.keySet()) {
+ if (nodeKey.startsWith(host + ":")) {
+ toClear.add(nodeKey);
+ }
+ }
+ toClear.forEach(lastUpdateTimes::remove);
+ }
+
private AmsServerInfo parseNodeKey(String nodeKey) {
String[] parts = nodeKey.split(":");
if (parts.length != 2) {
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/TestHighAvailabilityContainer.java
b/amoro-ams/src/test/java/org/apache/amoro/server/TestHighAvailabilityContainer.java
index 19048896a..7a8411ae8 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/TestHighAvailabilityContainer.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/TestHighAvailabilityContainer.java
@@ -84,7 +84,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testRegistAndElectWithoutMasterSlaveMode() throws Exception {
// Test that node registration is skipped when master-slave mode is
disabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, false);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
false);
haContainer = createContainerWithMockZk();
// Should not throw exception and should not register node
@@ -100,7 +100,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testRegistAndElectWithMasterSlaveMode() throws Exception {
// Test that node registration works when master-slave mode is enabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = createContainerWithMockZk();
// Register node
@@ -128,7 +128,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testGetAliveNodesWithoutMasterSlaveMode() throws Exception {
// Test that getAliveNodes returns empty list when master-slave mode is
disabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, false);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
false);
haContainer = createContainerWithMockZk();
List<AmsServerInfo> aliveNodes = haContainer.getAliveNodes();
@@ -142,7 +142,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testGetAliveNodesWhenNotLeader() throws Exception {
// Test that getAliveNodes returns empty list when not leader
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(false); // Not leader
haContainer = createContainerWithMockZk();
@@ -158,7 +158,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testGetAliveNodesAsLeader() throws Exception {
// Test that getAliveNodes returns nodes when leader
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(true); // Is leader
haContainer = createContainerWithMockZk();
@@ -185,7 +185,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testGetAliveNodesWithMultipleNodes() throws Exception {
// Test that getAliveNodes returns all registered nodes
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(true); // Is leader
haContainer = createContainerWithMockZk();
@@ -217,7 +217,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testCloseUnregistersNode() throws Exception {
// Test that close() unregisters the node
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = createContainerWithMockZk();
// Register node
@@ -240,7 +240,7 @@ public class TestHighAvailabilityContainer {
@Test
public void testHasLeadership() throws Exception {
// Test hasLeadership() method
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(false); // Not leader initially
haContainer = createContainerWithMockZk();
@@ -259,7 +259,7 @@ public class TestHighAvailabilityContainer {
public void testRegistAndElectWithoutHAEnabled() throws Exception {
// Test that registAndElect skips when HA is not enabled
serviceConfig.setBoolean(AmoroManagementConf.HA_ENABLE, false);
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = new ZkHighAvailabilityContainer(serviceConfig);
// Should not throw exception
@@ -309,7 +309,7 @@ public class TestHighAvailabilityContainer {
ZkHighAvailabilityContainer.class.getDeclaredField("isMasterSlaveMode");
isMasterSlaveModeField.setAccessible(true);
isMasterSlaveModeField.set(
- container,
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE));
+ container,
serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE));
if (serviceConfig.getBoolean(AmoroManagementConf.HA_ENABLE)) {
String haClusterName =
serviceConfig.getString(AmoroManagementConf.HA_CLUSTER_NAME);
diff --git
a/amoro-ams/src/test/java/org/apache/amoro/server/ha/TestZkHighAvailabilityContainer.java
b/amoro-ams/src/test/java/org/apache/amoro/server/ha/TestZkHighAvailabilityContainer.java
index 3f9433132..97b0d6d5c 100644
---
a/amoro-ams/src/test/java/org/apache/amoro/server/ha/TestZkHighAvailabilityContainer.java
+++
b/amoro-ams/src/test/java/org/apache/amoro/server/ha/TestZkHighAvailabilityContainer.java
@@ -83,7 +83,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testRegisterAndElectWithoutMasterSlaveMode() throws Exception {
// Test that node registration is skipped when master-slave mode is
disabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, false);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
false);
haContainer = createContainerWithMockZk();
// Should not throw exception and should not register node
@@ -99,7 +99,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testRegisterAndElectWithMasterSlaveMode() throws Exception {
// Test that node registration works when master-slave mode is enabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = createContainerWithMockZk();
// Register node
@@ -127,7 +127,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testGetAliveNodesWithoutMasterSlaveMode() throws Exception {
// Test that getAliveNodes returns empty list when master-slave mode is
disabled
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, false);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
false);
haContainer = createContainerWithMockZk();
List<AmsServerInfo> aliveNodes = haContainer.getAliveNodes();
@@ -141,7 +141,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testGetAliveNodesWhenNotLeader() throws Exception {
// Test that getAliveNodes returns empty list when not leader
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(false); // Not leader
haContainer = createContainerWithMockZk();
@@ -157,7 +157,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testGetAliveNodesAsLeader() throws Exception {
// Test that getAliveNodes returns nodes when leader
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(true); // Is leader
haContainer = createContainerWithMockZk();
@@ -184,7 +184,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testGetAliveNodesWithMultipleNodes() throws Exception {
// Test that getAliveNodes returns all registered nodes
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(true); // Is leader
haContainer = createContainerWithMockZk();
@@ -216,7 +216,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testCloseUnregistersNode() throws Exception {
// Test that close() unregisters the node
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = createContainerWithMockZk();
// Register node
@@ -239,7 +239,7 @@ public class TestZkHighAvailabilityContainer {
@Test
public void testHasLeadership() throws Exception {
// Test hasLeadership() method
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
mockLeaderLatch = createMockLeaderLatch(false); // Not leader initially
haContainer = createContainerWithMockZk();
@@ -258,7 +258,7 @@ public class TestZkHighAvailabilityContainer {
public void testRegisterAndElectWithoutHAEnabled() throws Exception {
// Test that registAndElect skips when HA is not enabled
serviceConfig.setBoolean(AmoroManagementConf.HA_ENABLE, false);
- serviceConfig.setBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE, true);
+ serviceConfig.setBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE,
true);
haContainer = new ZkHighAvailabilityContainer(serviceConfig);
// Should not throw exception
@@ -312,7 +312,7 @@ public class TestZkHighAvailabilityContainer {
ZkHighAvailabilityContainer.class.getDeclaredField("isMasterSlaveMode");
isMasterSlaveModeField.setAccessible(true);
isMasterSlaveModeField.set(
- container,
serviceConfig.getBoolean(AmoroManagementConf.USE_MASTER_SLAVE_MODE));
+ container,
serviceConfig.getBoolean(AmoroManagementConf.HA_USE_MASTER_SLAVE_MODE));
if (serviceConfig.getBoolean(AmoroManagementConf.HA_ENABLE)) {
String haClusterName =
serviceConfig.getString(AmoroManagementConf.HA_CLUSTER_NAME);
diff --git
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AbstractOptimizerOperator.java
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AbstractOptimizerOperator.java
index bda30c27f..af002bc0f 100644
---
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AbstractOptimizerOperator.java
+++
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AbstractOptimizerOperator.java
@@ -52,27 +52,58 @@ public class AbstractOptimizerOperator implements
Serializable {
private volatile boolean stopped = false;
private transient volatile AmsNodeManager amsNodeManager;
private transient volatile ThriftAmsNodeManager thriftAmsNodeManager;
+ // Throttle the "no AMS nodes discovered" warning to at most once per minute.
+ private volatile long lastEmptyNodesWarnTime = 0;
public AbstractOptimizerOperator(OptimizerConfig config) {
Preconditions.checkNotNull(config);
this.config = config;
- if (config.isMasterSlaveMode()) {
- String amsUrl = config.getAmsUrl();
- if (amsUrl.startsWith("zookeeper://")) {
- // ZK HA mode: discover nodes from ZooKeeper ephemeral nodes
- try {
- this.amsNodeManager = new AmsNodeManager(amsUrl);
- LOG.info("Initialized ZK AmsNodeManager for master-slave mode");
- } catch (Exception e) {
- LOG.warn("Failed to initialize AmsNodeManager, will use single AMS
URL", e);
+ // Node managers are initialized lazily in ensureNodeManagerInitialized()
rather than here,
+ // because Flink serializes operators on the client side and deserializes
them on TaskManagers.
+ // Since amsNodeManager and thriftAmsNodeManager are transient (they hold
non-serializable
+ // ZK/Thrift clients), they would be lost after deserialization. Lazy
initialization in
+ // getAmsNodeUrls() ensures they are properly created on the TaskManager
where they are needed.
+ }
+
+ /**
+ * Lazily initialize the node manager after Flink deserialization.
+ *
+ * <p>Flink serializes operators (including
OptimizerExecutor/OptimizerToucher) on the client side
+ * and deserializes them on TaskManagers. Since {@code amsNodeManager} and
{@code
+ * thriftAmsNodeManager} are {@code transient} (they hold non-serializable
ZK/Thrift clients),
+ * they are {@code null} after deserialization. This method re-creates them
on demand.
+ */
+ private void ensureNodeManagerInitialized() {
+ if (!config.isMasterSlaveMode()) {
+ return;
+ }
+ String amsUrl = config.getAmsUrl();
+ if (amsUrl.startsWith("zookeeper://")) {
+ if (amsNodeManager == null) {
+ synchronized (this) {
+ if (amsNodeManager == null) {
+ try {
+ this.amsNodeManager = new AmsNodeManager(amsUrl);
+ LOG.info("Initialized ZK AmsNodeManager for master-slave mode
(url: {})", amsUrl);
+ } catch (Exception e) {
+ LOG.warn("Failed to initialize AmsNodeManager, will use single
AMS URL", e);
+ }
+ }
}
- } else {
- // DB HA mode: discover nodes via getOptimizingNodeUrls() Thrift RPC
- try {
- this.thriftAmsNodeManager = new ThriftAmsNodeManager(amsUrl);
- LOG.info("Initialized ThriftAmsNodeManager for master-slave mode (DB
HA)");
- } catch (Exception e) {
- LOG.warn("Failed to initialize ThriftAmsNodeManager, will use single
AMS URL", e);
+ }
+ } else {
+ if (thriftAmsNodeManager == null) {
+ synchronized (this) {
+ if (thriftAmsNodeManager == null) {
+ try {
+ this.thriftAmsNodeManager = new ThriftAmsNodeManager(amsUrl);
+ LOG.info(
+ "Initialized ThriftAmsNodeManager for master-slave mode (DB
HA, url: {})",
+ amsUrl);
+ } catch (Exception e) {
+ LOG.warn("Failed to initialize ThriftAmsNodeManager, will use
single AMS URL", e);
+ }
+ }
}
}
}
@@ -80,11 +111,13 @@ public class AbstractOptimizerOperator implements
Serializable {
/** Get the AmsNodeManager instance if available (ZK mode). */
protected AmsNodeManager getAmsNodeManager() {
+ ensureNodeManagerInitialized();
return amsNodeManager;
}
/** Get the ThriftAmsNodeManager instance if available (DB mode). */
protected ThriftAmsNodeManager getThriftAmsNodeManager() {
+ ensureNodeManagerInitialized();
return thriftAmsNodeManager;
}
@@ -93,6 +126,7 @@ public class AbstractOptimizerOperator implements
Serializable {
* whether to use master-slave polling logic.
*/
protected boolean hasAmsNodeManager() {
+ ensureNodeManagerInitialized();
return amsNodeManager != null || thriftAmsNodeManager != null;
}
@@ -101,11 +135,22 @@ public class AbstractOptimizerOperator implements
Serializable {
* node manager is available or the node list is empty.
*/
protected List<String> getAmsNodeUrls() {
+ ensureNodeManagerInitialized();
if (amsNodeManager != null) {
List<String> urls = amsNodeManager.getAllAmsUrls();
if (!urls.isEmpty()) {
return urls;
}
+ long now = System.currentTimeMillis();
+ if (now - lastEmptyNodesWarnTime > TimeUnit.MINUTES.toMillis(1)) {
+ lastEmptyNodesWarnTime = now;
+ LOG.warn(
+ "Optimizer is configured for master-slave mode but no AMS nodes
were discovered "
+ + "from ZooKeeper nodes path. Falling back to the configured
URL [{}] (master only). "
+ + "Ensure that all AMS nodes have 'ha.use-master-slave-mode:
true' configured so "
+ + "they register themselves to ZooKeeper and can be discovered
by optimizers.",
+ getConfig().getAmsUrl());
+ }
}
if (thriftAmsNodeManager != null) {
return thriftAmsNodeManager.getAllAmsUrls();
@@ -174,10 +219,6 @@ public class AbstractOptimizerOperator implements
Serializable {
*/
protected <T> T callAuthenticatedAms(String amsUrl,
AmsAuthenticatedCallOperation<T> operation)
throws TException {
- // Maximum retry time window for auth errors in master-slave mode (30
seconds)
- long maxAuthRetryTimeWindow = TimeUnit.SECONDS.toMillis(30);
- Long firstAuthErrorTime = null;
-
// Per-node retry budget: in master-slave mode, limit consecutive
shouldRetryLater retries so
// that a permanently unreachable node does not block the multi-node
polling loop indefinitely.
int consecutiveRetries = 0;
@@ -190,50 +231,42 @@ public class AbstractOptimizerOperator implements
Serializable {
} catch (Throwable t) {
if (t instanceof AmoroException
&& ErrorCodes.PLUGIN_RETRY_AUTH_ERROR_CODE == ((AmoroException)
(t)).getErrorCode()) {
- // In master-slave mode, the slave node may not have completed
optimizer
- // auto-registration
- // yet, so we should retry within a time window before resetting
the token
boolean isMasterSlaveMode = config.isMasterSlaveMode() &&
hasAmsNodeManager();
- long currentTime = System.currentTimeMillis();
if (isMasterSlaveMode) {
- if (firstAuthErrorTime == null) {
- // First auth error, record the time
- firstAuthErrorTime = currentTime;
- }
-
- long elapsedTime = currentTime - firstAuthErrorTime;
- if (elapsedTime < maxAuthRetryTimeWindow) {
- // Still within retry time window, retry after waiting
- LOG.warn(
- "Got an authorization error while calling ams {} in
master-slave mode (elapsed: {}ms/{}ms). "
- + "This may be because the slave node hasn't completed
optimizer auto-registration yet. "
- + "Will retry after waiting.",
- amsUrl,
- elapsedTime,
- maxAuthRetryTimeWindow,
- t);
- waitAShortTime();
- } else {
- // Exceeded retry time window, reset token
- LOG.error(
- "Got authorization errors from ams {} in master-slave mode
for {}ms, exceeded retry time window ({}ms). "
- + "Reset token and wait for a new one",
- amsUrl,
- elapsedTime,
- maxAuthRetryTimeWindow,
- t);
- resetToken(token);
- firstAuthErrorTime = null;
+ // In master-slave mode, fail fast and let the multi-node
polling loop in
+ // OptimizerExecutor move on to the next node immediately.
+ //
+ // The root cause of auth failures on a slave is usually that
the slave has not yet
+ // synced the optimizer registration from the database. Blocking
here for 30 seconds
+ // and then resetting the GLOBAL token (which is also used by
the master and other
+ // healthy nodes) would:
+ // 1. Block the for-loop from polling other nodes for 30+
seconds.
+ // 2. Re-register with master → get new token → slave still
fails → infinite loop.
+ //
+ // Instead, throw immediately so pollTask() returns, the
for-loop advances to the
+ // next node, and the slave is retried on the NEXT poll round
(basePollInterval
+ // later).
+ // The global token is left intact so master/other nodes
continue to work normally.
+ LOG.warn(
+ "Optimizer executor got authorization error from AMS {} in
master-slave mode. "
+ + "The slave node may not have synced the optimizer
registration yet. "
+ + "Skipping this node for this round; will retry on the
next poll cycle.",
+ amsUrl,
+ t);
+ if (t instanceof TException) {
+ throw (TException) t;
}
+ throw new TException("Auth error for slave " + amsUrl, t);
} else {
- // Non-master-slave mode, reset token immediately
+ // Single-node mode: reset token and wait for OptimizerToucher
to re-register.
LOG.error(
- "Got a authorization error while calling ams {}, reset token
and wait for a new one",
+ "Got authorization error while calling AMS {} (token: {} is
now invalid). "
+ + "Token reset; will re-register on the next heartbeat
cycle.",
amsUrl,
+ token,
t);
resetToken(token);
- firstAuthErrorTime = null;
}
} else if (shouldReturnNull(t)) {
return null;
diff --git
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AmsNodeManager.java
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AmsNodeManager.java
index 94af2ca13..573e0cd87 100644
---
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AmsNodeManager.java
+++
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/AmsNodeManager.java
@@ -46,7 +46,7 @@ public class AmsNodeManager implements Serializable {
private final AtomicReference<List<String>> amsUrls =
new AtomicReference<>(Collections.emptyList());
private volatile long lastRefreshTime = 0;
- private transient Object refreshLock;
+ private transient volatile Object refreshLock;
public AmsNodeManager(String amsUrl) {
Matcher m = ZOOKEEPER_PATTERN.matcher(amsUrl);
@@ -59,6 +59,17 @@ public class AmsNodeManager implements Serializable {
refreshNodes();
}
+ private Object getRefreshLock() {
+ if (refreshLock == null) {
+ synchronized (this) {
+ if (refreshLock == null) {
+ refreshLock = new Object();
+ }
+ }
+ }
+ return refreshLock;
+ }
+
/** Get all available AMS URLs. */
public List<String> getAllAmsUrls() {
refreshNodesIfNeeded();
@@ -67,7 +78,7 @@ public class AmsNodeManager implements Serializable {
/** Manually refresh node list from ZooKeeper. */
public void refreshNodes() {
- synchronized (refreshLock) {
+ synchronized (getRefreshLock()) {
refreshNodesInternal();
lastRefreshTime = System.currentTimeMillis();
}
@@ -75,12 +86,11 @@ public class AmsNodeManager implements Serializable {
/** Refresh node list from ZooKeeper if needed. */
private void refreshNodesIfNeeded() {
- long now = System.currentTimeMillis();
- if (now - lastRefreshTime > REFRESH_INTERVAL_MS) {
- synchronized (refreshLock) {
- if (now - lastRefreshTime > REFRESH_INTERVAL_MS) {
+ if (System.currentTimeMillis() - lastRefreshTime > REFRESH_INTERVAL_MS) {
+ synchronized (getRefreshLock()) {
+ if (System.currentTimeMillis() - lastRefreshTime >
REFRESH_INTERVAL_MS) {
refreshNodesInternal();
- lastRefreshTime = now;
+ lastRefreshTime = System.currentTimeMillis();
}
}
}
@@ -89,24 +99,41 @@ public class AmsNodeManager implements Serializable {
/** Refresh node list from ZooKeeper (internal implementation). */
private void refreshNodesInternal() {
try {
- LOG.info("Refreshing nodes from {}", zkServerAddress);
+ LOG.debug("Refreshing AMS nodes from ZooKeeper: {}", zkServerAddress);
List<String> nodeUrls = new ArrayList<>();
List<AmsServerInfo> amsServerInfos =
AmsThriftUrl.parseMasterSlaveAmsNodes(zkServerAddress);
- LOG.info("Refreshing nodes from {}", amsServerInfos);
for (AmsServerInfo amsServerInfo : amsServerInfos) {
nodeUrls.add(buildAmsUrl(amsServerInfo));
}
+ List<String> previousUrls = amsUrls.get();
if (!nodeUrls.isEmpty()) {
amsUrls.set(nodeUrls);
- LOG.info("Refreshed AMS nodes, found {} nodes: {}", nodeUrls.size(),
nodeUrls);
+ if (!nodeUrls.equals(previousUrls)) {
+ LOG.info(
+ "AMS node list updated: {} -> {} ({} nodes)",
+ previousUrls.isEmpty() ? "(empty)" : previousUrls,
+ nodeUrls,
+ nodeUrls.size());
+ } else {
+ LOG.debug("AMS node list unchanged: {} ({} nodes)", nodeUrls,
nodeUrls.size());
+ }
} else {
- LOG.warn("No AMS nodes found in ZooKeeper");
amsUrls.set(Collections.emptyList());
+ if (!previousUrls.isEmpty()) {
+ LOG.warn(
+ "All AMS nodes disappeared from ZooKeeper (was: {}), "
+ + "will fall back to configured URL until nodes re-register",
+ previousUrls);
+ } else {
+ LOG.warn("No AMS nodes found in ZooKeeper path for {}",
zkServerAddress);
+ }
}
} catch (Exception e) {
- LOG.error("Failed to refresh AMS nodes from ZooKeeper", e);
- // Keep existing nodes on error
+ LOG.error(
+ "Failed to refresh AMS nodes from ZooKeeper, keeping previous list:
{}",
+ amsUrls.get(),
+ e);
}
}
diff --git
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
index f83da12da..26eda902c 100644
---
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
+++
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
@@ -34,6 +34,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -60,21 +61,67 @@ public class OptimizerExecutor extends
AbstractOptimizerOperator {
int consecutiveEmptyPolls = 0;
final int emptyPollThreshold = 5;
final Random random = new Random();
+ List<String> lastLoggedAmsUrls = Collections.emptyList();
+
+ LOG.info(
+ "Optimizer executor[{}] starting, mode: {}, configured AMS URL: {}",
+ threadId,
+ getConfig().isMasterSlaveMode() ? "master-slave" : "single-node",
+ getConfig().getAmsUrl());
while (isStarted()) {
List<String> amsUrls = getAmsNodeUrls();
+ // Log only when the node list changes, to avoid spamming on every poll
cycle.
+ if (!amsUrls.equals(lastLoggedAmsUrls)) {
+ LOG.info(
+ "Optimizer executor[{}] AMS node list changed: {} -> {}",
+ threadId,
+ lastLoggedAmsUrls.isEmpty() ? "(none)" : lastLoggedAmsUrls,
+ amsUrls.isEmpty() ? "(none)" : amsUrls);
+ lastLoggedAmsUrls = new ArrayList<>(amsUrls);
+ }
+
+ if (amsUrls.isEmpty()) {
+ // No AMS nodes discovered yet; warn periodically (first time and
every 12 cycles ≈ 1 min).
+ if (consecutiveEmptyPolls == 0 || consecutiveEmptyPolls % 12 == 0) {
+ LOG.warn(
+ "Optimizer executor[{}] no AMS nodes available (mode: {}, url:
{}), "
+ + "will retry in {}ms",
+ threadId,
+ getConfig().isMasterSlaveMode() ? "master-slave" : "single-node",
+ getConfig().getAmsUrl(),
+ basePollInterval);
+ }
+ consecutiveEmptyPolls++;
+ waitAShortTime(basePollInterval);
+ continue;
+ }
+
// Shuffle to prevent all optimizers from hitting the same AMS node
simultaneously
if (amsUrls.size() > 1) {
Collections.shuffle(amsUrls, random);
+ LOG.debug(
+ "Optimizer executor[{}] poll round starting, shuffled {} nodes:
{}",
+ threadId,
+ amsUrls.size(),
+ amsUrls);
}
boolean hasTask = false;
- for (String amsUrl : amsUrls) {
+ for (int i = 0; i < amsUrls.size(); i++) {
+ String amsUrl = amsUrls.get(i);
if (!isStarted()) {
break;
}
+ LOG.debug(
+ "Optimizer executor[{}] polling node [{}/{}]: {}",
+ threadId,
+ i + 1,
+ amsUrls.size(),
+ amsUrl);
+
OptimizingTask ackTask = null;
OptimizingTaskResult result = null;
try {
@@ -107,22 +154,41 @@ public class OptimizerExecutor extends
AbstractOptimizerOperator {
}
if (hasTask) {
+ if (consecutiveEmptyPolls >= emptyPollThreshold) {
+ LOG.info(
+ "Optimizer executor[{}] found task after {} consecutive empty
polls ({}), "
+ + "resuming normal poll interval",
+ threadId,
+ consecutiveEmptyPolls,
+ amsUrls);
+ }
consecutiveEmptyPolls = 0;
} else {
consecutiveEmptyPolls++;
}
long waitTime;
- if (amsUrls.isEmpty()) {
- waitTime = basePollInterval;
- } else if (consecutiveEmptyPolls >= emptyPollThreshold) {
+ if (consecutiveEmptyPolls >= emptyPollThreshold) {
int backoffFactor = consecutiveEmptyPolls - emptyPollThreshold + 1;
waitTime = Math.min(maxPollInterval, basePollInterval * (1L <<
backoffFactor));
- LOG.debug(
- "Optimizer executor[{}] no tasks found for {} consecutive polls,
using increased interval: {}ms",
- threadId,
- consecutiveEmptyPolls,
- waitTime);
+ if (consecutiveEmptyPolls == emptyPollThreshold) {
+ LOG.info(
+ "Optimizer executor[{}] no tasks found for {} consecutive polls
across {} node(s) {}, "
+ + "slowing down poll interval to {}ms (max: {}ms)",
+ threadId,
+ consecutiveEmptyPolls,
+ amsUrls.size(),
+ amsUrls,
+ waitTime,
+ maxPollInterval);
+ } else {
+ LOG.debug(
+ "Optimizer executor[{}] no tasks found for {} consecutive polls,
"
+ + "using increased poll interval: {}ms",
+ threadId,
+ consecutiveEmptyPolls,
+ waitTime);
+ }
} else {
waitTime = basePollInterval;
}
diff --git
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerToucher.java
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerToucher.java
index 48b8a7e07..9fd24e22e 100644
---
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerToucher.java
+++
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerToucher.java
@@ -69,6 +69,13 @@ public class OptimizerToucher extends
AbstractOptimizerOperator {
private boolean checkToken() {
if (!tokenIsReady()) {
+ LOG.info(
+ "Registering optimizer to AMS {} (group: {}, mode: {}, threads: {},
memory: {}MB) ...",
+ getConfig().getAmsUrl(),
+ getConfig().getGroupName(),
+ getConfig().isMasterSlaveMode() ? "master-slave" : "single-node",
+ getConfig().getExecutionParallel(),
+ getConfig().getMemorySize());
try {
String token =
callAms(
@@ -89,10 +96,18 @@ public class OptimizerToucher extends
AbstractOptimizerOperator {
if (tokenChangeListener != null) {
tokenChangeListener.tokenChange(token);
}
- LOG.info("Registered optimizer to ams with token:{}", token);
+ LOG.info(
+ "Successfully registered optimizer to AMS {} (group: {}) with
token: {}",
+ getConfig().getAmsUrl(),
+ getConfig().getGroupName(),
+ token);
return true;
} catch (TException e) {
- LOG.error("Register optimizer to ams failed", e);
+ LOG.error(
+ "Failed to register optimizer to AMS {} (group: {})",
+ getConfig().getAmsUrl(),
+ getConfig().getGroupName(),
+ e);
if (e instanceof AmoroException
&& ErrorCodes.FORBIDDEN_ERROR_CODE == ((AmoroException)
e).getErrorCode()) {
System.exit(1); // Don't need to try again
@@ -104,20 +119,27 @@ public class OptimizerToucher extends
AbstractOptimizerOperator {
}
private void touch() {
+ String currentToken = getToken();
try {
callAms(
client -> {
- client.touch(getToken());
+ client.touch(currentToken);
return null;
});
- LOG.debug("Optimizer[{}] touch ams", getToken());
+ LOG.debug(
+ "Optimizer touch AMS {} succeeded (token: {})",
getConfig().getAmsUrl(), currentToken);
} catch (TException e) {
if (e instanceof AmoroException
&& ErrorCodes.PLUGIN_RETRY_AUTH_ERROR_CODE == ((AmoroException)
e).getErrorCode()) {
+ LOG.error(
+ "Got authorization error from AMS {} during touch (token: {} is
now invalid). "
+ + "Will re-register on the next heartbeat cycle.",
+ getConfig().getAmsUrl(),
+ currentToken,
+ e);
setToken(null);
- LOG.error("Got authorization error from ams, try to register later",
e);
} else {
- LOG.error("Touch ams failed", e);
+ LOG.error("Touch AMS {} failed (token: {})", getConfig().getAmsUrl(),
currentToken, e);
}
}
}
diff --git
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/ThriftAmsNodeManager.java
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/ThriftAmsNodeManager.java
index f325d5c6e..f4373f28f 100644
---
a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/ThriftAmsNodeManager.java
+++
b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/ThriftAmsNodeManager.java
@@ -43,7 +43,7 @@ public class ThriftAmsNodeManager implements Serializable {
private final AtomicReference<List<String>> amsUrls =
new AtomicReference<>(Collections.emptyList());
private volatile long lastRefreshTime = 0;
- private transient Object refreshLock;
+ private transient volatile Object refreshLock;
public ThriftAmsNodeManager(String initialAmsUrl) {
this.initialAmsUrl = initialAmsUrl;
@@ -70,12 +70,11 @@ public class ThriftAmsNodeManager implements Serializable {
}
private void refreshNodesIfNeeded() {
- long now = System.currentTimeMillis();
- if (now - lastRefreshTime > REFRESH_INTERVAL_MS) {
+ if (System.currentTimeMillis() - lastRefreshTime > REFRESH_INTERVAL_MS) {
synchronized (getLock()) {
- if (now - lastRefreshTime > REFRESH_INTERVAL_MS) {
+ if (System.currentTimeMillis() - lastRefreshTime >
REFRESH_INTERVAL_MS) {
refreshNodesInternal();
- lastRefreshTime = now;
+ lastRefreshTime = System.currentTimeMillis();
}
}
}
@@ -103,7 +102,11 @@ public class ThriftAmsNodeManager implements Serializable {
private Object getLock() {
if (refreshLock == null) {
- refreshLock = new Object();
+ synchronized (this) {
+ if (refreshLock == null) {
+ refreshLock = new Object();
+ }
+ }
}
return refreshLock;
}
diff --git a/docs/configuration/ams-config.md b/docs/configuration/ams-config.md
index 3285e57e6..c00841dae 100644
--- a/docs/configuration/ams-config.md
+++ b/docs/configuration/ams-config.md
@@ -81,6 +81,7 @@ table td:last-child, table th:last-child { width: 40%;
word-break: break-all; }
| ha.node-offline.timeout | 5 min | Timeout duration to determine if a node is
offline. After this duration, the node's bucket IDs will be reassigned. |
| ha.session-timeout | 30 s | The Zookeeper session timeout in milliseconds. |
| ha.type | zk | High availability implementation type: zk or database. |
+| ha.use-master-slave-mode | false | This setting controls whether to enable
the AMS horizontal scaling feature, which is currently under development and
testing. |
| ha.zookeeper-address | | The Zookeeper address used for high availability. |
| ha.zookeeper-auth-keytab | | The Zookeeper authentication keytab file path
when auth type is KERBEROS. |
| ha.zookeeper-auth-principal | | The Zookeeper authentication principal when
auth type is KERBEROS. |
@@ -145,7 +146,6 @@ table td:last-child, table th:last-child { width: 40%;
word-break: break-all; }
| thrift-server.selector-thread-count | 2 | The number of selector threads for
the Thrift server. |
| thrift-server.table-service.bind-port | 1260 | Port that the table service
thrift server is bound to. |
| thrift-server.table-service.worker-thread-count | 20 | The number of worker
threads for the Thrift server. |
-| use-master-slave-mode | false | This setting controls whether to enable the
AMS horizontal scaling feature, which is currently under development and
testing. |
## RBAC Example
