This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 8d9545ea585 feat: Define clone sync criteria in
CoordinatorDynamicConfig (#20247)
8d9545ea585 is described below
commit 8d9545ea585a5d49cdd27384b4a3b365f1e30dd7
Author: Kashif Faraz <[email protected]>
AuthorDate: Thu Sep 10 10:37:08 2026 +0530
feat: Define clone sync criteria in CoordinatorDynamicConfig (#20247)
When cloning historicals for a blue-green deployment using
`CoordinatorDynamicConfig.cloneServers`,
it can often be difficult to determine if the clones are already synced and
can be safely terminated.
This is primarily because the source server keeps getting assigned new
segments and the clone target
is always playing catch up.
The metric that has been used so far to determine if a clone is
sufficiently "synced" is the
`segmentLoadsRemaining` returned by the `/cloneStatus` API, but that number
often fails to fall below
any reasonable threshold thereby slowing down the deployment.
Changes
---------
- Add a configurable `CloneSyncCriteria` to `CoordinatorDynamicConfig`
- The criteria is a function of segments "pending sync" during a
coordinator run i.e. segments which
are already loaded on the source server but still loading at the target
server at the start of the run.
- If the criteria is met, update the state of the clone to `SYNCED` and
emit a metric
- In subsequent runs, it is possible for the state to go back to
`IN_PROGRESS` if the clone starts lagging
behind again.
- Add new metrics to track number of segments pending sync and the synced
status of a clone
---
docs/api-reference/dynamic-configuration-api.md | 7 +-
docs/configuration/index.md | 3 +
docs/operations/metrics.md | 5 +
.../org/apache/druid/common/config/Configs.java | 9 ++
.../apache/druid/common/config/ConfigsTest.java | 5 +-
.../server/coordinator/CloneSyncCriteria.java | 75 ++++++++++++
.../coordinator/CoordinatorDynamicConfig.java | 29 +++++
.../server/coordinator/ServerCloneStatus.java | 134 ++++-----------------
.../server/coordinator/duty/CloneHistoricals.java | 105 +++++++++++++++-
.../druid/server/coordinator/stats/Stats.java | 4 +
.../CoordinatorDynamicConfigTest.java | 6 +-
.../server/coordinator/ServerCloneStatusTest.java | 11 +-
.../simulate/CoordinatorSimulation.java | 8 ++
.../simulate/CoordinatorSimulationBaseTest.java | 9 ++
.../simulate/CoordinatorSimulationBuilder.java | 9 ++
.../simulate/HistoricalCloningTest.java | 61 ++++++++++
.../CoordinatorDynamicConfigsResourceTest.java | 2 +-
17 files changed, 359 insertions(+), 123 deletions(-)
diff --git a/docs/api-reference/dynamic-configuration-api.md
b/docs/api-reference/dynamic-configuration-api.md
index fefbc22c313..94c8c52beb3 100644
--- a/docs/api-reference/dynamic-configuration-api.md
+++ b/docs/api-reference/dynamic-configuration-api.md
@@ -108,6 +108,7 @@ Host: http://ROUTER_IP:ROUTER_PORT
"debugDimensions": null,
"turboLoadingNodes": [],
"cloneServers": {},
+ "cloneSyncCriteria": {},
"historicalTierAliases": {
"hot": ["hot_1", "hot_2"]
}
@@ -180,7 +181,8 @@ curl
"http://ROUTER_IP:ROUTER_PORT/druid/coordinator/v1/config"; \
"maxNonPrimaryReplicantsToLoad": 2147483647,
"useRoundRobinSegmentAssignment": true,
"turboLoadingNodes": [],
- "cloneServers": {}
+ "cloneServers": {},
+ "cloneSyncCriteria": {}
}'
```
@@ -213,7 +215,8 @@ Content-Length: 683
"maxNonPrimaryReplicantsToLoad": 2147483647,
"useRoundRobinSegmentAssignment": true,
"turboLoadingNodes": [],
- "cloneServers": {}
+ "cloneServers": {},
+ "cloneSyncCriteria": {}
}
```
diff --git a/docs/configuration/index.md b/docs/configuration/index.md
index c4def258739..d07756b858e 100644
--- a/docs/configuration/index.md
+++ b/docs/configuration/index.md
@@ -805,6 +805,9 @@ The following table shows the dynamic configuration
properties for the Coordinat
|`replicateAfterLoadTimeout`|Boolean flag for whether or not additional
replication is needed for segments that have failed to load due to the expiry
of `druid.coordinator.load.timeout`. If this is set to true, the Coordinator
will attempt to replicate the failed segment on a different historical server.
This helps improve the segment availability if there are a few slow Historicals
in the cluster. However, the slow Historical may still load the segment later
and the Coordinator may issu [...]
|`turboLoadingNodes`| Experimental. List of Historical servers to place in
turbo loading mode. These servers use a larger thread-pool to load segments
faster but at the cost of query performance. For servers specified in
`turboLoadingNodes`, `druid.coordinator.loadqueuepeon.http.batchSize` is
ignored and the coordinator uses the value of the respective
`numLoadingThreads` instead.<br/>Please use this config with caution. All
servers should eventually be removed from this list once the se [...]
|`cloneServers`| Experimental. Map from target Historical server to source
Historical server which should be cloned by the target. The target Historical
does not participate in regular segment assignment or balancing. Instead, the
Coordinator mirrors any segment assignment made to the source Historical onto
the target Historical, so that the target becomes an exact copy of the source.
Segments on the target Historical do not count towards replica counts either.
If the source disappears, [...]
+|`cloneSyncCriteria`| Experimental. Criteria defining when a clone historical
should be considered as "synced" to its source server. The criteria is a
function of the number and percentage of segments "pending sync", i.e. segments
already loaded on the source server but still loading on the target server.
Segments which are yet to be loaded on the source server itself do not affect
the sync status.|See fields below.|
+|`cloneSyncCriteria.maxSegmentsPendingSync`| Experimental. This is a sub-field
inside `cloneSyncCriteria`. For a clone to be considered "synced" with its
source server, the number of segments pending sync must be less than or equal
to this value. |100|
+|`cloneSyncCriteria.maxPercentPendingSync` | Experimental. This is a sub-field
inside `cloneSyncCriteria`. For a clone to be considered "synced" with its
source server, the percentage of segments pending sync must be less than or
equal to this value. |1|
|`historicalTierAliases`|Map from a virtual tier name to the set of real
Historical tier names it expands to. When a load/drop rule references a virtual
alias tier, the Coordinator replaces it with its real tiers — each receiving
the full replica count independently. The alias key itself is never loaded to
directly. For example, `{"hot": ["hot_1", "hot_2"]}` causes a rule of `{"hot":
2}` to load 2 replicas on each of `hot_1` and `hot_2`; `hot` receives no direct
assignment. An alias valu [...]
##### Smart segment loading
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 8cad874d1a4..62d01316e6a 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -452,6 +452,9 @@ These metrics are emitted by the Druid Coordinator in every
run of the correspon
|`segment/assignSkipped/count`|Number of segments that could not be assigned
to any server for loading. This can occur due to replication throttling, no
available disk space, or a full load queue.|`dataSource`, `server`, `tier`,
`description`|Varies|
|`segment/moveSkipped/count`|Number of segments that were chosen for balancing
but could not be moved. This can occur when segments are already optimally
placed.|`dataSource`, `server`, `tier`, `description`|Varies|
|`segment/dropSkipped/count`|Number of segments that could not be dropped from
any server.|`dataSource`, `server`, `tier`, `description`|Varies|
+|`segment/clone/assigned/count`|Number of segments assigned to be loaded on a
historical clone.|`dataSource`, `server`, `tier`|Varies|
+|`segment/clone/dropped/count`|Number of segments dropped from a historical
clone.|`dataSource`, `server`, `tier`|Varies|
+|`segment/clone/pendingSync/count`|Number of segments that still need to be
loaded on a historical clone but are already loaded on its source
server.|`server`, `tier`|Varies|
|`segment/loadQueue/size`|Size in bytes of segments to load.|`server`|Varies|
|`segment/loadQueue/count`|Number of segments to load.|`server`|Varies|
|`segment/loading/rateKbps`|Current rate of segment loading on a server in
kbps (1000 bits per second). The rate is calculated as a moving average over
the last 10 GiB or more of successful segment loads on that
server.|`server`|Varies|
@@ -468,6 +471,8 @@ These metrics are emitted by the Druid Coordinator in every
run of the correspon
|`segment/underReplicated/count`|Number of segments, including replicas, left
to load until all used segments are available for queries.|`tier`,
`dataSource`|0|
|`segment/availableDeepStorageOnly/count`|Number of unique segments that are
only available for querying directly from deep storage.|`dataSource`|Varies|
|`tier/historical/count`|Number of available historical nodes in each tier.
The `tierAlias` dimension is emitted only when the tier belongs to an alias
configured via
[`historicalTierAliases`](../configuration/index.md#dynamic-configuration), and
can be used to aggregate metrics across the tiers in an alias.|`tier`,
`tierAlias`|Varies|
+|`tier/historical/clone/count`|Number of historical nodes in a tier which are
a clone of another historical in the same or different tier. The `tierAlias`
dimension is emitted only when the tier belongs to an alias configured via
[`historicalTierAliases`](../configuration/index.md#dynamic-configuration), and
can be used to aggregate metrics across the tiers in an alias.|`tier`,
`tierAlias`|Varies|
+|`tier/historical/clone/synced`|Number of historical clones in a tier which
are currently in sync with their source server. |`server`, `tier`|1 if synced,
O if not synced|
|`tier/replication/factor`|Configured maximum replication factor in each tier.
The `tierAlias` dimension is emitted only when the tier belongs to an alias
configured via
[`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`,
`tierAlias`|Varies|
|`tier/required/capacity`|Total capacity in bytes required in each tier. The
`tierAlias` dimension is emitted only when the tier belongs to an alias
configured via
[`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`,
`tierAlias`|Varies|
|`tier/total/capacity`|Total capacity in bytes available in each tier. The
`tierAlias` dimension is emitted only when the tier belongs to an alias
configured via
[`historicalTierAliases`](../configuration/index.md#dynamic-configuration).|`tier`,
`tierAlias`|Varies|
diff --git
a/processing/src/main/java/org/apache/druid/common/config/Configs.java
b/processing/src/main/java/org/apache/druid/common/config/Configs.java
index 433a2548a4b..d2d10ea804c 100644
--- a/processing/src/main/java/org/apache/druid/common/config/Configs.java
+++ b/processing/src/main/java/org/apache/druid/common/config/Configs.java
@@ -42,6 +42,15 @@ public class Configs
return value == null ? defaultValue : value;
}
+ /**
+ * Returns the given {@code value} if it is not null, otherwise returns the
+ * {@code defaultValue}.
+ */
+ public static double valueOrDefault(Double value, double defaultValue)
+ {
+ return value == null ? defaultValue : value;
+ }
+
/**
* Returns the given {@code value} if it is not null, otherwise returns the
* {@code defaultValue}.
diff --git
a/processing/src/test/java/org/apache/druid/common/config/ConfigsTest.java
b/processing/src/test/java/org/apache/druid/common/config/ConfigsTest.java
index ff4acf7cbea..770aef077da 100644
--- a/processing/src/test/java/org/apache/druid/common/config/ConfigsTest.java
+++ b/processing/src/test/java/org/apache/druid/common/config/ConfigsTest.java
@@ -31,7 +31,10 @@ public class ConfigsTest
Assertions.assertEquals(11, Configs.valueOrDefault((Integer) null, 11));
Assertions.assertEquals(10, Configs.valueOrDefault((Long) 10L, 11L));
- Assertions.assertEquals(11, Configs.valueOrDefault(null, 11L));
+ Assertions.assertEquals(11, Configs.valueOrDefault((Long) null, 11L));
+
+ Assertions.assertEquals(10.0, Configs.valueOrDefault((Double) 10.0, 11.0));
+ Assertions.assertEquals(11.0, Configs.valueOrDefault(null, 11.0));
Assertions.assertFalse(Configs.valueOrDefault((Boolean) false, true));
Assertions.assertTrue(Configs.valueOrDefault(null, true));
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/CloneSyncCriteria.java
b/server/src/main/java/org/apache/druid/server/coordinator/CloneSyncCriteria.java
new file mode 100644
index 00000000000..682378c3f31
--- /dev/null
+++
b/server/src/main/java/org/apache/druid/server/coordinator/CloneSyncCriteria.java
@@ -0,0 +1,75 @@
+/*
+ * 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.druid.server.coordinator;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.error.InvalidInput;
+
+import javax.annotation.Nullable;
+
+/**
+ * Criteria defining when a clone historical should be considered as
+ * {@link ServerCloneStatus.State#SYNCED} to its source server. The criteria is
+ * a function of the number and percentage of segments "pending sync", i.e.
+ * segments already loaded on the source server but still loading on the target
+ * server. Segments which are yet to be loaded on the source server itself do
not
+ * affect the sync status.
+ */
+public record CloneSyncCriteria(
+ @JsonProperty("maxSegmentsPendingSync") @Nullable Integer
maxSegmentsPendingSync,
+ @JsonProperty("maxPercentPendingSync") @Nullable Double
maxPercentPendingSync
+)
+{
+ public static final int DEFAULT_MAX_SEGMENTS_PENDING_SYNC = 100;
+ public static final double DEFAULT_MAX_PERCENT_PENDING_SYNC = 1.0;
+
+ public CloneSyncCriteria
+ {
+ InvalidInput.conditionalException(
+ maxSegmentsPendingSync == null || maxSegmentsPendingSync >= 0,
+ "'maxSegmentsPendingSync' must be greater than or equal to 0"
+ );
+ InvalidInput.conditionalException(
+ maxPercentPendingSync == null || (maxPercentPendingSync >= 0.0 &&
maxPercentPendingSync <= 100.0),
+ "'maxPercentPendingSync' must be in the range [0.0, 100.0]"
+ );
+ }
+
+ /**
+ * For a clone to be considered SYNCED, the number of segments pending sync
+ * must be less than or equal to this value.
+ * Default value is {@link #DEFAULT_MAX_SEGMENTS_PENDING_SYNC}.
+ */
+ public int getMaxSegmentsPendingSync()
+ {
+ return Configs.valueOrDefault(maxSegmentsPendingSync,
DEFAULT_MAX_SEGMENTS_PENDING_SYNC);
+ }
+
+ /**
+ * For a clone to be considered SYNCED, the percentage of segments pending
sync
+ * must be less than or equal to this value.
+ * Default value is {@link #DEFAULT_MAX_PERCENT_PENDING_SYNC}.
+ */
+ public double getMaxPercentPendingSync()
+ {
+ return Configs.valueOrDefault(maxPercentPendingSync,
DEFAULT_MAX_PERCENT_PENDING_SYNC);
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfig.java
b/server/src/main/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfig.java
index e0ea80e2446..554a8219294 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfig.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfig.java
@@ -77,6 +77,7 @@ public class CoordinatorDynamicConfig
private final Set<String> turboLoadingNodes;
private final Map<String, String> cloneServers;
+ private final CloneSyncCriteria cloneSyncCriteria;
/**
* Map from alias tier name to the set of actual tier names it represents.
@@ -143,6 +144,7 @@ public class CoordinatorDynamicConfig
@JsonProperty("debugDimensions") @Nullable Map<String, String>
debugDimensions,
@JsonProperty("turboLoadingNodes") @Nullable Set<String>
turboLoadingNodes,
@JsonProperty("cloneServers") @Nullable Map<String, String> cloneServers,
+ @JsonProperty("cloneSyncCriteria") @Nullable CloneSyncCriteria
cloneSyncCriteria,
@JsonProperty("historicalTierAliases") @Nullable Map<String,
Set<String>> historicalTierAliases
)
{
@@ -189,6 +191,7 @@ public class CoordinatorDynamicConfig
this.validDebugDimensions = validateDebugDimensions(debugDimensions);
this.turboLoadingNodes = Configs.valueOrDefault(turboLoadingNodes,
Set.of());
this.cloneServers = Configs.valueOrDefault(cloneServers, Map.of());
+ this.cloneSyncCriteria = Configs.valueOrDefault(cloneSyncCriteria, new
CloneSyncCriteria(null, null));
this.historicalTierAliases = Configs.valueOrDefault(historicalTierAliases,
Map.of());
final Set<String> aliasKeys = this.historicalTierAliases.keySet();
@@ -392,6 +395,18 @@ public class CoordinatorDynamicConfig
return cloneServers;
}
+ /**
+ * Criteria defining when a target server specified in {@link #cloneServers}
+ * can be considered as {@link ServerCloneStatus.State#SYNCED}. A clone once
+ * marked as synced may again go out of sync in subsequent coordinator cycles
+ * if any of the conditions or sync criteria change.
+ */
+ @JsonProperty
+ public CloneSyncCriteria getCloneSyncCriteria()
+ {
+ return cloneSyncCriteria;
+ }
+
@JsonProperty
public Map<String, Set<String>> getHistoricalTierAliases()
{
@@ -442,6 +457,7 @@ public class CoordinatorDynamicConfig
", replicateAfterLoadTimeout=" + replicateAfterLoadTimeout +
", turboLoadingNodes=" + turboLoadingNodes +
", cloneServers=" + cloneServers +
+ ", cloneSyncCriteria=" + cloneSyncCriteria +
", historicalTierAliases=" + historicalTierAliases +
'}';
}
@@ -480,6 +496,7 @@ public class CoordinatorDynamicConfig
&& Objects.equals(turboLoadingNodes, that.turboLoadingNodes)
&& Objects.equals(debugDimensions, that.debugDimensions)
&& Objects.equals(cloneServers, that.cloneServers)
+ && Objects.equals(cloneSyncCriteria, that.cloneSyncCriteria)
&& Objects.equals(historicalTierAliases,
that.historicalTierAliases);
}
@@ -505,6 +522,7 @@ public class CoordinatorDynamicConfig
debugDimensions,
turboLoadingNodes,
cloneServers,
+ cloneSyncCriteria,
historicalTierAliases
);
}
@@ -562,6 +580,7 @@ public class CoordinatorDynamicConfig
private Boolean smartSegmentLoading;
private Set<String> turboLoadingNodes;
private Map<String, String> cloneServers;
+ private CloneSyncCriteria cloneSyncCriteria;
private Map<String, Set<String>> historicalTierAliases;
public Builder()
@@ -588,6 +607,7 @@ public class CoordinatorDynamicConfig
@JsonProperty("debugDimensions") @Nullable Map<String, String>
debugDimensions,
@JsonProperty("turboLoadingNodes") @Nullable Set<String>
turboLoadingNodes,
@JsonProperty("cloneServers") @Nullable Map<String, String>
cloneServers,
+ @JsonProperty("cloneSyncCriteria") @Nullable CloneSyncCriteria
cloneSyncCriteria,
@JsonProperty("historicalTierAliases") @Nullable Map<String,
Set<String>> historicalTierAliases
)
{
@@ -609,6 +629,7 @@ public class CoordinatorDynamicConfig
this.debugDimensions = debugDimensions;
this.turboLoadingNodes = turboLoadingNodes;
this.cloneServers = cloneServers;
+ this.cloneSyncCriteria = cloneSyncCriteria;
this.historicalTierAliases = historicalTierAliases;
}
@@ -708,6 +729,12 @@ public class CoordinatorDynamicConfig
return this;
}
+ public Builder withCloneSyncCriteria(CloneSyncCriteria cloneSyncCriteria)
+ {
+ this.cloneSyncCriteria = cloneSyncCriteria;
+ return this;
+ }
+
public Builder withHistoricalTierAliases(Map<String, Set<String>>
historicalTierAliases)
{
this.historicalTierAliases = historicalTierAliases;
@@ -742,6 +769,7 @@ public class CoordinatorDynamicConfig
debugDimensions,
turboLoadingNodes,
cloneServers,
+ cloneSyncCriteria,
historicalTierAliases
);
}
@@ -775,6 +803,7 @@ public class CoordinatorDynamicConfig
valueOrDefault(debugDimensions, defaults.getDebugDimensions()),
valueOrDefault(turboLoadingNodes, defaults.getTurboLoadingNodes()),
valueOrDefault(cloneServers, defaults.getCloneServers()),
+ valueOrDefault(cloneSyncCriteria, defaults.getCloneSyncCriteria()),
valueOrDefault(historicalTierAliases,
defaults.getHistoricalTierAliases())
);
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/ServerCloneStatus.java
b/server/src/main/java/org/apache/druid/server/coordinator/ServerCloneStatus.java
index 0d011fd1243..56e319438be 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/ServerCloneStatus.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/ServerCloneStatus.java
@@ -19,127 +19,31 @@
package org.apache.druid.server.coordinator;
-import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
-import java.util.Objects;
-
/**
* Immutable class which represents the current status of a single clone
server.
+ *
+ * @param segmentsPendingSync Number of segments that are already loaded on the
+ * source server but are yet to be loaded on the
target server.
*/
-public class ServerCloneStatus
+public record ServerCloneStatus(
+ @JsonProperty("sourceServer") String sourceServer,
+ @JsonProperty("targetServer") String targetServer,
+ @JsonProperty("state") State state,
+ @JsonProperty("segmentLoadsRemaining") long segmentLoadsRemaining,
+ @JsonProperty("segmentDropsRemaining") long segmentDropsRemaining,
+ @JsonProperty("segmentsPendingSync") long segmentsPendingSync,
+ @JsonProperty("percentPendingSync") double percentPendingSync,
+ @JsonProperty("bytesToLoad") long bytesToLoad
+)
{
- private final String sourceServer;
- private final String targetServer;
- private final State state;
- private final long segmentLoadsRemaining;
- private final long segmentDropsRemaining;
- private final long bytesToLoad;
-
- @JsonCreator
- public ServerCloneStatus(
- @JsonProperty("sourceServer") String sourceServer,
- @JsonProperty("targetServer") String targetServer,
- @JsonProperty("state") State state,
- @JsonProperty("segmentLoadsRemaining") long segmentLoadsRemaining,
- @JsonProperty("segmentDropsRemaining") long segmentDropsRemaining,
- @JsonProperty("bytesToLoad") long bytesToLoad
- )
- {
- this.sourceServer = sourceServer;
- this.targetServer = targetServer;
- this.state = state;
- this.segmentLoadsRemaining = segmentLoadsRemaining;
- this.segmentDropsRemaining = segmentDropsRemaining;
- this.bytesToLoad = bytesToLoad;
- }
-
- @JsonProperty
- public String getSourceServer()
- {
- return sourceServer;
- }
-
- @JsonProperty
- public String getTargetServer()
- {
- return targetServer;
- }
-
- @JsonProperty
- public long getSegmentLoadsRemaining()
- {
- return segmentLoadsRemaining;
- }
-
- @JsonProperty
- public long getSegmentDropsRemaining()
- {
- return segmentDropsRemaining;
- }
-
- @JsonProperty
- public long getBytesToLoad()
- {
- return bytesToLoad;
- }
-
- @JsonProperty
- public State getState()
- {
- return state;
- }
-
/**
* Create a {@link ServerCloneStatus} where the current status is unknown as
the target server is missing.
*/
public static ServerCloneStatus unknown(String sourceServer, String
targetServer)
{
- return new ServerCloneStatus(sourceServer, targetServer,
State.TARGET_SERVER_MISSING, -1, -1, -1);
- }
-
- @Override
- public boolean equals(Object o)
- {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- ServerCloneStatus that = (ServerCloneStatus) o;
- return segmentLoadsRemaining == that.segmentLoadsRemaining
- && segmentDropsRemaining == that.segmentDropsRemaining
- && bytesToLoad == that.bytesToLoad
- && Objects.equals(sourceServer, that.sourceServer)
- && Objects.equals(targetServer, that.targetServer)
- && state == that.state;
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(
- sourceServer,
- targetServer,
- state,
- segmentLoadsRemaining,
- segmentDropsRemaining,
- bytesToLoad
- );
- }
-
- @Override
- public String toString()
- {
- return "ServerCloneStatus{" +
- "sourceServer='" + sourceServer + '\'' +
- ", targetServer='" + targetServer + '\'' +
- ", state=" + state +
- ", segmentLoadsRemaining=" + segmentLoadsRemaining +
- ", segmentDropsRemaining=" + segmentDropsRemaining +
- ", bytesToLoad=" + bytesToLoad +
- '}';
+ return new ServerCloneStatus(sourceServer, targetServer,
State.TARGET_SERVER_MISSING, -1, -1, -1, -1, -1);
}
/**
@@ -159,6 +63,14 @@ public class ServerCloneStatus
/**
* Segments are loaded or being loaded. The counts give a better view of
the progress.
*/
- IN_PROGRESS
+ IN_PROGRESS,
+ /**
+ * Clone server has caught up with the source server. This is not a
permanent
+ * state and if sync criteria are updated or if the clone starts lagging
behind,
+ * subsequent coordinator cycles may move the clone back to {@link
#IN_PROGRESS}.
+ *
+ * @see CloneSyncCriteria
+ */
+ SYNCED
}
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/duty/CloneHistoricals.java
b/server/src/main/java/org/apache/druid/server/coordinator/duty/CloneHistoricals.java
index 82441810634..5112f61a3c1 100644
---
a/server/src/main/java/org/apache/druid/server/coordinator/duty/CloneHistoricals.java
+++
b/server/src/main/java/org/apache/druid/server/coordinator/duty/CloneHistoricals.java
@@ -22,6 +22,7 @@ package org.apache.druid.server.coordinator.duty;
import org.apache.druid.client.ImmutableDruidDataSource;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.server.coordinator.CloneStatusManager;
+import org.apache.druid.server.coordinator.CloneSyncCriteria;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.server.coordinator.DruidCluster;
import org.apache.druid.server.coordinator.DruidCoordinatorRuntimeParams;
@@ -30,6 +31,7 @@ import org.apache.druid.server.coordinator.ServerHolder;
import org.apache.druid.server.coordinator.loading.PartialLoadProfile;
import org.apache.druid.server.coordinator.loading.SegmentAction;
import org.apache.druid.server.coordinator.loading.SegmentLoadQueueManager;
+import org.apache.druid.server.coordinator.stats.CoordinatorRunStats;
import org.apache.druid.server.coordinator.stats.Dimension;
import org.apache.druid.server.coordinator.stats.RowKey;
import org.apache.druid.server.coordinator.stats.Stats;
@@ -94,6 +96,7 @@ public class CloneHistoricals implements CoordinatorDuty
serverHolder -> serverHolder
));
+ final Map<String, CloningStats> targetHistoricalStats = new HashMap<>();
for (Map.Entry<String, String> entry : cloneServers.entrySet()) {
final String targetHistoricalName = entry.getKey();
final ServerHolder targetServer =
hostToHistoricalMap.get(targetHistoricalName);
@@ -110,6 +113,9 @@ public class CloneHistoricals implements CoordinatorDuty
continue;
}
+ final CloningStats cloningStats = new
CloningStats(sourceServer.getServer().getNumSegments());
+ targetHistoricalStats.put(targetHistoricalName, cloningStats);
+
final Set<DataSegment> sourceProjectedSegments =
sourceServer.getProjectedSegments();
final Set<DataSegment> targetProjectedSegments =
targetServer.getProjectedSegments();
// Load any segment that the clone target is missing, or that it holds
under a different partial-load profile
@@ -120,6 +126,12 @@ public class CloneHistoricals implements CoordinatorDuty
if (shouldLoadSegmentOnTargetServer(segment, sourceProfile,
targetServer, targetProjectedSegments)) {
loadSegmentOnTargetServer(segment, sourceProfile, targetServer,
params);
}
+
+ if (params.isUsedSegment(segment)
+ && targetServer.isLoadingSegment(segment)
+ && sourceServer.isServingSegment(segment)) {
+
cloningStats.incrementMissingSegmentCount(sourceServer.isServingSegment(segment));
+ }
}
// Drop any segments missing from the clone source.
@@ -130,7 +142,12 @@ public class CloneHistoricals implements CoordinatorDuty
}
}
- final Map<String, ServerCloneStatus> newStatusMap =
createCurrentStatusMap(hostToHistoricalMap, cloneServers);
+ final Map<String, ServerCloneStatus> newStatusMap = createCurrentStatusMap(
+ hostToHistoricalMap,
+ cloneServers,
+ targetHistoricalStats,
+ params
+ );
cloneStatusManager.updateStatus(newStatusMap);
return params;
@@ -149,6 +166,7 @@ public class CloneHistoricals implements CoordinatorDuty
{
final RowKey.Builder rowKey = RowKey
.with(Dimension.SERVER, targetServer.getServer().getName())
+ .with(Dimension.TIER, targetServer.getServer().getTier())
.with(Dimension.DATASOURCE, segment.getDataSource());
final DataSegment loadableSegment = getLoadableSegment(segment, params);
@@ -187,9 +205,13 @@ public class CloneHistoricals implements CoordinatorDuty
if (targetServer.isLoadingSegment(segment)) {
targetServer.cancelLoad(segment);
} else if (loadQueueManager.dropSegment(segment, targetServer)) {
+ final RowKey rowKey = RowKey
+ .with(Dimension.SERVER, targetServer.getServer().getName())
+ .with(Dimension.TIER, targetServer.getServer().getTier())
+ .and(Dimension.DATASOURCE, segment.getDataSource());
params.getCoordinatorStats().add(
Stats.Segments.DROPPED_FROM_CLONE,
- RowKey.of(Dimension.SERVER, targetServer.getServer().getName()),
+ rowKey,
1L
);
}
@@ -221,10 +243,13 @@ public class CloneHistoricals implements CoordinatorDuty
*/
private Map<String, ServerCloneStatus> createCurrentStatusMap(
Map<String, ServerHolder> historicalMap,
- Map<String, String> cloneServers
+ Map<String, String> cloneServers,
+ Map<String, CloningStats> targetHistoricalStats,
+ DruidCoordinatorRuntimeParams params
)
{
final Map<String, ServerCloneStatus> newStatusMap = new HashMap<>();
+ final CloneSyncCriteria cloneSyncCriteria =
params.getCoordinatorDynamicConfig().getCloneSyncCriteria();
for (Map.Entry<String, String> entry : cloneServers.entrySet()) {
final String targetServerName = entry.getKey();
@@ -239,10 +264,12 @@ public class CloneHistoricals implements CoordinatorDuty
if (targetServer == null) {
newStatus = ServerCloneStatus.unknown(sourceServerName,
targetServerName);
} else {
-
ServerCloneStatus.State state;
+ final CloningStats stats =
targetHistoricalStats.getOrDefault(targetServerName, new CloningStats(0));
if (!historicalMap.containsKey(sourceServerName)) {
state = ServerCloneStatus.State.SOURCE_SERVER_MISSING;
+ } else if (isSynced(stats, cloneSyncCriteria)) {
+ state = ServerCloneStatus.State.SYNCED;
} else {
state = ServerCloneStatus.State.IN_PROGRESS;
}
@@ -255,7 +282,17 @@ public class CloneHistoricals implements CoordinatorDuty
segmentDrop += 1;
}
}
- newStatus = new ServerCloneStatus(sourceServerName, targetServerName,
state, segmentLoad, segmentDrop, bytesLeft);
+ newStatus = new ServerCloneStatus(
+ sourceServerName,
+ targetServerName,
+ state,
+ segmentLoad,
+ segmentDrop,
+ stats.segmentsPendingSync,
+ stats.percentPendingSync(),
+ bytesLeft
+ );
+ collectMetrics(targetServer, newStatus, params.getCoordinatorStats());
}
newStatusMap.put(targetServerName, newStatus);
}
@@ -285,4 +322,62 @@ public class CloneHistoricals implements CoordinatorDuty
final PartialLoadProfile targetProfile =
targetServer.getProjectedProfile(segment);
return !Objects.equals(fingerprintOf(sourceProfile),
fingerprintOf(targetProfile));
}
+
+ private boolean isSynced(CloningStats stats, CloneSyncCriteria criteria)
+ {
+ return stats.segmentsPendingSync <= criteria.getMaxSegmentsPendingSync()
+ && stats.percentPendingSync() <= criteria.getMaxPercentPendingSync();
+ }
+
+ /**
+ * Adds metrics for this clone to the run stats.
+ */
+ private void collectMetrics(
+ ServerHolder targetServer,
+ ServerCloneStatus newStatus,
+ CoordinatorRunStats stats
+ )
+ {
+ final String targetServerName = targetServer.getServer().getName();
+ final RowKey rowKey = RowKey.with(Dimension.SERVER, targetServerName)
+ .and(Dimension.TIER,
targetServer.getServer().getTier());
+ stats.add(Stats.Segments.PENDING_SYNC_ON_CLONE, rowKey,
newStatus.segmentsPendingSync());
+
+ final boolean isSynced = newStatus.state() ==
ServerCloneStatus.State.SYNCED;
+ stats.add(Stats.Tier.CLONE_SYNCED, rowKey, isSynced ? 1L : 0L);
+ }
+
+ private static class CloningStats
+ {
+ final int totalSegmentsLoadedOnSource;
+
+ /**
+ * Number of segments already loaded on source server but not on target
server.
+ */
+ int segmentsPendingSync = 0;
+
+ CloningStats(int totalSegmentsLoadedOnSource)
+ {
+ this.totalSegmentsLoadedOnSource = totalSegmentsLoadedOnSource;
+ }
+
+ void incrementMissingSegmentCount(boolean loadedOnSourceServer)
+ {
+ if (loadedOnSourceServer) {
+ ++segmentsPendingSync;
+ }
+ }
+
+ /**
+ * Percentage of segments pending sync.
+ */
+ double percentPendingSync()
+ {
+ if (totalSegmentsLoadedOnSource > 0) {
+ return (100.0 * segmentsPendingSync) / totalSegmentsLoadedOnSource;
+ } else {
+ return 0.0;
+ }
+ }
+ }
}
diff --git
a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
index 80368ba7558..138d7068189 100644
--- a/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
+++ b/server/src/main/java/org/apache/druid/server/coordinator/stats/Stats.java
@@ -71,6 +71,8 @@ public class Stats
= CoordinatorStat.toDebugAndEmit("cloneLoad",
"segment/clone/assigned/count");
public static final CoordinatorStat DROPPED_FROM_CLONE
= CoordinatorStat.toDebugAndEmit("cloneDrop",
"segment/clone/dropped/count");
+ public static final CoordinatorStat PENDING_SYNC_ON_CLONE
+ = CoordinatorStat.toDebugAndEmit("clonePendingSync",
"segment/clone/pendingSync/count");
// Partial-load reconciliation in a run
public static final CoordinatorStat PARTIAL_ASSIGNED
@@ -118,6 +120,8 @@ public class Stats
= CoordinatorStat.toDebugAndEmit("numHistorical",
"tier/historical/count");
public static final CoordinatorStat CLONE_COUNT
= CoordinatorStat.toDebugAndEmit("numClones",
"tier/historical/clone/count");
+ public static final CoordinatorStat CLONE_SYNCED
+ = CoordinatorStat.toDebugAndEmit("cloneSynced",
"tier/historical/clone/synced");
}
public static class Compaction
diff --git
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfigTest.java
similarity index 99%
rename from
server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigTest.java
rename to
server/src/test/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfigTest.java
index 95b436fc745..421ca427a36 100644
---
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/CoordinatorDynamicConfigTest.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.druid.server.http;
+package org.apache.druid.server.coordinator;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
@@ -25,7 +25,7 @@ import com.google.common.collect.ImmutableSet;
import nl.jqno.equalsverifier.EqualsVerifier;
import org.apache.druid.error.DruidException;
import org.apache.druid.segment.TestHelper;
-import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.apache.druid.server.http.SegmentLoadingMode;
import org.apache.druid.utils.JvmUtils;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -281,6 +281,7 @@ public class CoordinatorDynamicConfigTest
null,
ImmutableSet.of("host1"),
null,
+ null,
null
);
Assertions.assertTrue(config.getSpecificDataSourcesToKillUnusedSegmentsIn().isEmpty());
@@ -308,6 +309,7 @@ public class CoordinatorDynamicConfigTest
null,
ImmutableSet.of("host1"),
null,
+ null,
null
);
Assertions.assertEquals(ImmutableSet.of("test1"),
config.getSpecificDataSourcesToKillUnusedSegmentsIn());
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/ServerCloneStatusTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/ServerCloneStatusTest.java
index e0aa95e493f..6172d9b9c9e 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/ServerCloneStatusTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/ServerCloneStatusTest.java
@@ -29,7 +29,16 @@ public class ServerCloneStatusTest
@Test
public void testSerde() throws Exception
{
- ServerCloneStatus metrics = new ServerCloneStatus("host2", "host1",
ServerCloneStatus.State.IN_PROGRESS, 3012, 10, 100);
+ ServerCloneStatus metrics = new ServerCloneStatus(
+ "host2",
+ "host1",
+ ServerCloneStatus.State.IN_PROGRESS,
+ 3012,
+ 10,
+ 1,
+ 1,
+ 100
+ );
byte[] bytes = DefaultObjectMapper.INSTANCE.writeValueAsBytes(metrics);
ServerCloneStatus deserialized =
DefaultObjectMapper.INSTANCE.readValue(bytes, ServerCloneStatus.class);
Assertions.assertEquals(deserialized, metrics);
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulation.java
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulation.java
index 8a6ff4b8205..b1b6a618ff4 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulation.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulation.java
@@ -22,10 +22,12 @@ package org.apache.druid.server.coordinator.simulate;
import org.apache.druid.client.DruidServer;
import org.apache.druid.java.util.metrics.MetricsVerifier;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.apache.druid.server.coordinator.ServerCloneStatus;
import org.apache.druid.server.coordinator.loading.SegmentHolder;
import org.apache.druid.server.coordinator.rules.Rule;
import org.apache.druid.timeline.DataSegment;
+import javax.annotation.Nullable;
import java.util.List;
/**
@@ -97,6 +99,12 @@ public interface CoordinatorSimulation
* Gets the load percentage of the specified datasource as seen by the
coordinator.
*/
double getLoadPercentage(String datasource);
+
+ /**
+ * Gets the current cloning status for the given clone target server.
+ */
+ @Nullable
+ ServerCloneStatus getCloneStatus(DruidServer cloneTarget);
}
interface ClusterState
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
index 9e8e16c8f25..59d4bb73321 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBaseTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.segment.TestDataSource;
import org.apache.druid.server.coordination.ServerType;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.server.coordinator.CreateDataSegments;
+import org.apache.druid.server.coordinator.ServerCloneStatus;
import org.apache.druid.server.coordinator.loading.SegmentHolder;
import
org.apache.druid.server.coordinator.rules.ForeverBroadcastDistributionRule;
import org.apache.druid.server.coordinator.rules.ForeverDropRule;
@@ -37,6 +38,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
+import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -169,6 +171,13 @@ public abstract class CoordinatorSimulationBaseTest
implements
return sim.coordinator().getLoadPercentage(datasource);
}
+ @Nullable
+ @Override
+ public ServerCloneStatus getCloneStatus(DruidServer cloneTarget)
+ {
+ return sim.coordinator().getCloneStatus(cloneTarget);
+ }
+
// Verification methods
void verifyDatasourceIsFullyLoaded(String datasource)
{
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
index 3484b7782d9..5088d450cbd 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/CoordinatorSimulationBuilder.java
@@ -48,6 +48,7 @@ import
org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
import org.apache.druid.server.coordinator.DruidCompactionConfig;
import org.apache.druid.server.coordinator.DruidCoordinator;
import org.apache.druid.server.coordinator.MetadataManager;
+import org.apache.druid.server.coordinator.ServerCloneStatus;
import org.apache.druid.server.coordinator.balancer.BalancerStrategyFactory;
import
org.apache.druid.server.coordinator.balancer.CachingCostBalancerStrategyConfig;
import
org.apache.druid.server.coordinator.balancer.CachingCostBalancerStrategyFactory;
@@ -73,6 +74,7 @@ import org.apache.druid.timeline.SegmentId;
import org.easymock.EasyMock;
import org.joda.time.Duration;
+import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -416,6 +418,13 @@ public class CoordinatorSimulationBuilder
return coordinator.getDatasourceToLoadStatus().get(datasource);
}
+ @Nullable
+ @Override
+ public ServerCloneStatus getCloneStatus(DruidServer cloneTarget)
+ {
+ return env.cloneStatusManager.getStatusForServer(cloneTarget.getName());
+ }
+
@Override
public MetricsVerifier getMetricsVerifier()
{
diff --git
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/HistoricalCloningTest.java
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/HistoricalCloningTest.java
index 1eedb33363d..9ee48912677 100644
---
a/server/src/test/java/org/apache/druid/server/coordinator/simulate/HistoricalCloningTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordinator/simulate/HistoricalCloningTest.java
@@ -21,12 +21,16 @@ package org.apache.druid.server.coordinator.simulate;
import org.apache.druid.client.DruidServer;
import org.apache.druid.segment.TestDataSource;
+import org.apache.druid.server.coordinator.CloneSyncCriteria;
import org.apache.druid.server.coordinator.CoordinatorDynamicConfig;
+import org.apache.druid.server.coordinator.ServerCloneStatus;
import org.apache.druid.server.coordinator.stats.Stats;
+import org.apache.druid.timeline.DataSegment;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.List;
import java.util.Map;
public class HistoricalCloningTest extends CoordinatorSimulationBaseTest
@@ -344,4 +348,61 @@ public class HistoricalCloningTest extends
CoordinatorSimulationBaseTest
Assertions.assertEquals(5, historicalT11.getTotalSegments());
Assertions.assertEquals(5, historicalT12.getTotalSegments());
}
+
+ @Test
+ public void test_cloneIsSynced_ifLoadedSegmentsMeetCriteria()
+ {
+ final List<DataSegment> segments = Segments.WIKI_10X1D;
+ final CoordinatorSimulation sim =
+ CoordinatorSimulation.builder()
+ .withSegments(segments)
+ .withServers(historicalT11, historicalT12)
+ .withRules(datasource, Load.on(Tier.T1,
1).forever())
+ .withDynamicConfig(
+ CoordinatorDynamicConfig
+ .builder()
+
.withCloneServers(Map.of(historicalT12.getHost(), historicalT11.getHost()))
+ .withCloneSyncCriteria(new
CloneSyncCriteria(1, 15.0))
+ .withSmartSegmentLoading(true)
+ .build()
+ )
+ .build();
+
+
+ // Load 9 segments on histT11
+ segments.forEach(historicalT11::addDataSegment);
+ historicalT11.removeDataSegment(segments.getFirst().getId());
+
+ // Load 8 segments on histT12
+ segments.forEach(historicalT12::addDataSegment);
+ historicalT12.removeDataSegment(segments.getFirst().getId());
+ historicalT12.removeDataSegment(segments.getLast().getId());
+
+ startSimulation(sim);
+ runCoordinatorCycle();
+
+ // Verify that 2 segments are assigned to the clone but only 1 is required
for full sync
+ verifyValue(
+ Stats.Segments.ASSIGNED_TO_CLONE.getMetricName(),
+ Map.of("server", historicalT12.getName()),
+ 2L
+ );
+ verifyValue(
+ Stats.Segments.PENDING_SYNC_ON_CLONE.getMetricName(),
+ Map.of("server", historicalT12.getName()),
+ 1L
+ );
+ verifyValue(
+ Stats.Tier.CLONE_SYNCED.getMetricName(),
+ Map.of("server", historicalT12.getName(), "tier", Tier.T1),
+ 1L
+ );
+
+ // Verify that the clone is already considered as synced since criteria is
met
+ final ServerCloneStatus cloneStatus = getCloneStatus(historicalT12);
+ Assertions.assertNotNull(cloneStatus);
+ Assertions.assertEquals(2, cloneStatus.segmentLoadsRemaining());
+ Assertions.assertEquals(1, cloneStatus.segmentsPendingSync());
+ Assertions.assertEquals(ServerCloneStatus.State.SYNCED,
cloneStatus.state());
+ }
}
diff --git
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigsResourceTest.java
b/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigsResourceTest.java
index b1ba0f1b36e..d2ca54b74ef 100644
---
a/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigsResourceTest.java
+++
b/server/src/test/java/org/apache/druid/server/http/CoordinatorDynamicConfigsResourceTest.java
@@ -84,7 +84,7 @@ public class CoordinatorDynamicConfigsResourceTest
public void testGetCloneStatus()
{
List<ServerCloneStatus> statusMetrics = ImmutableList.of(
- new ServerCloneStatus("hist3", "hist1",
ServerCloneStatus.State.IN_PROGRESS, 2, 0, 1000),
+ new ServerCloneStatus("hist3", "hist1",
ServerCloneStatus.State.IN_PROGRESS, 2, 0, 1, 1, 1000),
ServerCloneStatus.unknown("hist4", "hist3")
);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
