This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 6c8f8c04b37 SegmentStatusChecker batch read the zk segment metadata
instead of per segment (#19252)
6c8f8c04b37 is described below
commit 6c8f8c04b37fb334c30735ccd06e068c0541d733
Author: Jhow <[email protected]>
AuthorDate: Wed Aug 19 05:17:36 2026 +0800
SegmentStatusChecker batch read the zk segment metadata instead of per
segment (#19252)
---
.../pinot/common/metadata/ZKMetadataProvider.java | 31 +-
.../controller/helix/SegmentStatusChecker.java | 57 +++-
.../helix/core/PinotHelixResourceManager.java | 9 +
.../controller/helix/SegmentStatusCheckerTest.java | 370 +++++++++++++++++----
.../PinotHelixResourceManagerStatelessTest.java | 46 +++
5 files changed, 438 insertions(+), 75 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
index dc8a91a01b3..b2a9da428ba 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
@@ -20,6 +20,7 @@ package org.apache.pinot.common.metadata;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
@@ -639,8 +640,36 @@ public class ZKMetadataProvider {
return getTableSchema(propertyStore, tableConfig.getTableName());
}
+ /// Reads the ZK metadata of the named segments of the given table in a
single batched request, and returns it
+ /// index-aligned with `segmentNames`. An entry is `null` when the segment's
znode could not be read, either because
+ /// it does not exist or because the read failed; the two are not
distinguished.
+ ///
+ /// When `stats` is non-null it is filled with the [Stat] of each segment's
znode, also index-aligned with
+ /// `segmentNames`, and `null` wherever the record is `null`.
+ public static List<SegmentZKMetadata>
getSegmentsZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
+ String tableNameWithType, List<String> segmentNames, @Nullable
List<Stat> stats) {
+ int numSegments = segmentNames.size();
+ List<String> paths = new ArrayList<>(numSegments);
+ for (String segmentName : segmentNames) {
+ paths.add(constructPropertyStorePathForSegment(tableNameWithType,
segmentName));
+ }
+ List<ZNRecord> znRecords = propertyStore.get(paths, stats,
AccessOption.PERSISTENT, false);
+ Preconditions.checkState(znRecords.size() == numSegments,
+ "Got %s segment ZN records for %s segments of table: %s",
znRecords.size(), numSegments, tableNameWithType);
+ Preconditions.checkState(stats == null || stats.size() == numSegments,
+ "Got %s znode stats for %s segments of table: %s", stats != null ?
stats.size() : 0, numSegments,
+ tableNameWithType);
+ List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>(numSegments);
+ for (ZNRecord znRecord : znRecords) {
+ segmentsZKMetadata.add(znRecord != null ? new
SegmentZKMetadata(znRecord) : null);
+ }
+ return segmentsZKMetadata;
+ }
+
/// NOTE: this method is very expensive, use
[#getSegments(ZkHelixPropertyStore, String)] instead if only
- /// segment names are needed.
+ /// segment names are needed. Segments whose ZK metadata cannot be read are
dropped from the returned list; use
+ /// [#getSegmentsZKMetadata(ZkHelixPropertyStore, String, List, List)] when
the result must line up with a known list
+ /// of segments, or when the znodes' [Stat] is needed.
public static List<SegmentZKMetadata>
getSegmentsZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
String tableNameWithType) {
String parentPath =
constructPropertyStorePathForResource(tableNameWithType);
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
index 4ed16098a86..cb27b48a9d7 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.helix;
+import com.google.common.annotations.VisibleForTesting;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
@@ -78,6 +79,8 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
// log messages about disabled tables at most once a day
private static final long DISABLED_TABLE_LOG_INTERVAL_MS =
TimeUnit.DAYS.toMillis(1);
private static final int MAX_SEGMENTS_TO_LOG = 10;
+ // Number of segments whose ZK metadata is read per batched request, see
updateSegmentMetrics()
+ private static final int SEGMENT_METADATA_BATCH_SIZE = 10_000;
private final int _waitForPushTimeSeconds;
private final TableSizeReader _tableSizeReader;
@@ -88,6 +91,9 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
private final Map<String, Set<String>> _tableTenantMap = new HashMap<>();
private long _lastDisabledTableLogTimestamp = 0;
+ // Overridden by the tests so that the batching can be exercised without a
table larger than the batch size
+ @VisibleForTesting
+ int _segmentMetadataBatchSize = SEGMENT_METADATA_BATCH_SIZE;
/// Constructs the segment status checker.
/// @param pinotHelixResourceManager The resource checker used to interact
with Helix
@@ -382,10 +388,36 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
List<String> unavailableSegmentsByState = new ArrayList<>();
List<String> unavailableSegmentsByInstance = new ArrayList<>();
- for (String segment : segments) {
+ // The segment ZK metadata and the znode stats are read with one batched
request per SEGMENT_METADATA_BATCH_SIZE
+ // segments instead of one request per segment. The batch is bounded
because the metadata of a whole batch is held
+ // in heap while it is being checked, which does not scale to tables with
hundreds of thousands of segments.
+ List<String> segmentsToCheck = new ArrayList<>(segments);
+ List<SegmentZKMetadata> batchSegmentsZKMetadata = List.of();
+ List<Stat> batchSegmentStats = List.of();
+ int batchStartIndex = 0;
+ // Number of segments whose ZK metadata was read back, tracked over all
the batches so that a table whose metadata
+ // could not be read at all is told apart from a table that is genuinely
unhealthy
+ int numSegmentsWithZKMetadata = 0;
+
+ for (int i = 0; i < numSegments; i++) {
+ // The batches are aligned to multiples of the batch size, so a new one
starts exactly on these indexes
+ if (i % _segmentMetadataBatchSize == 0) {
+ batchStartIndex = i;
+ int batchEndIndex = Math.min(i + _segmentMetadataBatchSize,
numSegments);
+ batchSegmentStats = new ArrayList<>(batchEndIndex - batchStartIndex);
+ batchSegmentsZKMetadata =
_pinotHelixResourceManager.getSegmentsZKMetadata(tableNameWithType,
+ segmentsToCheck.subList(batchStartIndex, batchEndIndex),
batchSegmentStats);
+ for (SegmentZKMetadata segmentZKMetadata : batchSegmentsZKMetadata) {
+ if (segmentZKMetadata != null) {
+ numSegmentsWithZKMetadata++;
+ }
+ }
+ }
+ String segment = segmentsToCheck.get(i);
+ Map<String, String> isStateMap = idealState.getInstanceStateMap(segment);
// Number of replicas in ideal state that is in ONLINE/CONSUMING state
int numISReplicasUp = 0;
- for (Map.Entry<String, String> entry :
idealState.getInstanceStateMap(segment).entrySet()) {
+ for (Map.Entry<String, String> entry : isStateMap.entrySet()) {
String state = entry.getValue();
if (state.equals(SegmentStateModel.ONLINE) ||
state.equals(SegmentStateModel.CONSUMING)) {
numISReplicasUp++;
@@ -397,7 +429,7 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
}
maxISReplicasUp = Math.max(maxISReplicasUp, numISReplicasUp);
- SegmentZKMetadata segmentZKMetadata =
_pinotHelixResourceManager.getSegmentZKMetadata(tableNameWithType, segment);
+ SegmentZKMetadata segmentZKMetadata = batchSegmentsZKMetadata.get(i -
batchStartIndex);
// Skip the segment when it doesn't have ZK metadata. Most likely the
segment is just deleted.
if (segmentZKMetadata == null) {
segmentsWithoutZKMetadata.add(segment);
@@ -419,11 +451,10 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
// The grace window is _waitForPushTimeSeconds. Once a segment is
older than it and still
// under-replicated, it is checked normally, so genuinely stuck
commits and real replica losses still alert.
// The comparison uses evSnapshotTimestamp instead of
System.currentTimeMillis() because for large tables
- // with many segments, the status check can take several minutes.
A segment updated after
- // the EV snapshot was taken but before this individual segment
check runs could be incorrectly flagged as
- // OFFLINE when using current time.
- Stat segmentStat = propertyStore == null ? null : propertyStore.getStat(
-
ZKMetadataProvider.constructPropertyStorePathForSegment(tableNameWithType,
segment), AccessOption.PERSISTENT);
+ // with many segments, the status check can still take a while. A
segment updated after the EV snapshot was
+ // taken but before this individual segment check runs could be
incorrectly flagged as OFFLINE when using
+ // current time.
+ Stat segmentStat = batchSegmentStats.get(i - batchStartIndex);
long refTimeMs = segmentStat != null ? segmentStat.getMtime() :
segmentZKMetadata.getCreationTime();
if (refTimeMs > evSnapshotTimestamp - _waitForPushTimeSeconds * 1000L) {
continue;
@@ -472,10 +503,18 @@ public class SegmentStatusChecker extends
ControllerPeriodicTask<SegmentStatusCh
minEVReplicasUp = Math.min(minEVReplicasUp, numEVReplicasUp);
// Total number of replicas in ideal state (including ERROR/OFFLINE
states)
- int numISReplicasTotal =
Math.max(idealState.getInstanceStateMap(segment).entrySet().size(), 1);
+ int numISReplicasTotal = Math.max(isStateMap.size(), 1);
minEVReplicasUpPercent = Math.min(minEVReplicasUpPercent,
numEVReplicasUp * 100 / numISReplicasTotal);
}
+ // Not a single segment's ZK metadata could be read, so the gauges
computed above describe nothing. Leave the
+ // table's gauges alone instead of publishing all-green values that would
silence the alerts a stale gauge fires.
+ if (numSegmentsWithZKMetadata == 0) {
+ LOGGER.error("Failed to read the ZK metadata of all {} segments of
table: {}, skipping the metric update",
+ numSegments, tableNameWithType);
+ return false;
+ }
+
// Log unavailable segments in batches
if (!unavailableSegmentsByState.isEmpty()) {
LOGGER.warn("Table {} has {} segments marked unavailable due to
non-ONLINE/CONSUMING states: {}",
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 70703680f7a..06af564d024 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -1056,6 +1056,15 @@ public class PinotHelixResourceManager {
return ZKMetadataProvider.getSegmentsZKMetadata(_propertyStore,
tableNameWithType);
}
+ /// Reads the ZK metadata of the named segments in a single batched request,
index-aligned with `segmentNames` and
+ /// holding `null` for segments whose znode could not be read. See
+ /// [ZKMetadataProvider#getSegmentsZKMetadata(ZkHelixPropertyStore, String,
List, List)] for the full contract,
+ /// including the `stats` out-parameter that carries the znodes'
modification times.
+ public List<SegmentZKMetadata> getSegmentsZKMetadata(String
tableNameWithType,
+ List<String> segmentNames, @Nullable List<Stat> stats) {
+ return ZKMetadataProvider.getSegmentsZKMetadata(_propertyStore,
tableNameWithType, segmentNames, stats);
+ }
+
public Collection<String> getLastLLCCompletedSegments(String
tableNameWithType) {
return
getLastLLCCompletedSegments(getSegmentsZKMetadata(tableNameWithType));
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
index b5d390d0215..02e333eb52c 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
@@ -21,6 +21,8 @@ package org.apache.pinot.controller.helix;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -58,6 +60,7 @@ import org.apache.pinot.spi.utils.TimeUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.zookeeper.data.Stat;
+import org.mockito.ArgumentCaptor;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -68,10 +71,12 @@ import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
@SuppressWarnings("unchecked")
@@ -125,7 +130,7 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata segmentZKMetadata = mockPushedSegmentZKMetadata(1234,
11111L);
- when(resourceManager.getSegmentZKMetadata(eq(OFFLINE_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, OFFLINE_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -142,6 +147,10 @@ public class SegmentStatusCheckerTest {
runSegmentStatusChecker(resourceManager, 0);
verifyControllerMetrics(OFFLINE_TABLE_NAME, 2, 5, 3, 2, 66, 1, 100, 2,
2468);
+ // The metadata and the znode stats must come from batched reads: a
per-segment read here costs two blocking
+ // ZooKeeper round trips per segment, which takes minutes on tables with
hundreds of thousands of segments.
+ verify(resourceManager, never()).getSegmentZKMetadata(anyString(),
anyString());
+ verify(propertyStore, never()).getStat(anyString(), anyInt());
}
private SegmentZKMetadata mockPushedSegmentZKMetadata(long sizeInBytes, long
pushTimeMs) {
@@ -158,13 +167,10 @@ public class SegmentStatusCheckerTest {
private void runSegmentStatusChecker(PinotHelixResourceManager
resourceManager, int waitForPushTimeInSeconds,
TableSizeReader tableSizeReader) {
- LeadControllerManager leadControllerManager =
mock(LeadControllerManager.class);
- when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
- ControllerConf controllerConf = mock(ControllerConf.class);
-
when(controllerConf.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(waitForPushTimeInSeconds);
- SegmentStatusChecker segmentStatusChecker =
- new SegmentStatusChecker(resourceManager, leadControllerManager,
controllerConf, _controllerMetrics,
- tableSizeReader);
+ runSegmentStatusChecker(buildSegmentStatusChecker(resourceManager,
waitForPushTimeInSeconds, tableSizeReader));
+ }
+
+ private void runSegmentStatusChecker(SegmentStatusChecker
segmentStatusChecker) {
segmentStatusChecker.start();
segmentStatusChecker.run();
}
@@ -297,10 +303,9 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committedSegmentZKMetadata =
mockCommittedSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg1)).thenReturn(committedSegmentZKMetadata);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg2)).thenReturn(committedSegmentZKMetadata);
SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg3)).thenReturn(consumingSegmentZKMetadata);
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of(seg1, committedSegmentZKMetadata, seg2,
committedSegmentZKMetadata, seg3, consumingSegmentZKMetadata));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -363,10 +368,9 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committedSegmentZKMetadata =
mockCommittedSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg1)).thenReturn(committedSegmentZKMetadata);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg2)).thenReturn(committedSegmentZKMetadata);
SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg3)).thenReturn(consumingSegmentZKMetadata);
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of(seg1, committedSegmentZKMetadata, seg2,
committedSegmentZKMetadata, seg3, consumingSegmentZKMetadata));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -436,10 +440,9 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committedSegmentZKMetadata =
mockCommittedSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg1)).thenReturn(committedSegmentZKMetadata);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg2)).thenReturn(committedSegmentZKMetadata);
SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg3)).thenReturn(consumingSegmentZKMetadata);
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of(seg1, committedSegmentZKMetadata, seg2,
committedSegmentZKMetadata, seg3, consumingSegmentZKMetadata));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -522,10 +525,9 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committedSegmentZKMetadata =
mockCommittedSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg1)).thenReturn(committedSegmentZKMetadata);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg2)).thenReturn(committedSegmentZKMetadata);
SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg3)).thenReturn(consumingSegmentZKMetadata);
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of(seg1, committedSegmentZKMetadata, seg2,
committedSegmentZKMetadata, seg3, consumingSegmentZKMetadata));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -563,7 +565,7 @@ public class SegmentStatusCheckerTest {
// A pauseless COMMITTING segment: done consuming but its immutable segment
is still being built/loaded on the
// replicas. The grace check keys off the segment znode's modification time
(mtime), which the test drives via
- // propertyStore.getStat(...); the metadata here only supplies status and
size.
+ // mockSegmentsZKMetadata(...); the metadata here only supplies status and
size.
private SegmentZKMetadata mockCommittingSegmentZKMetadata() {
SegmentZKMetadata segmentZKMetadata = mock(SegmentZKMetadata.class);
when(segmentZKMetadata.getStatus()).thenReturn(Status.COMMITTING);
@@ -578,6 +580,48 @@ public class SegmentStatusCheckerTest {
return stat;
}
+ /// Stubs the single batched segment read with the metadata of
`segmentZKMetadataMap` (segment name -> metadata) and
+ /// the znode modification times of `segmentZNodeMTimesMs` (segment name ->
mtime in epoch millis), preserving the
+ /// index alignment with the requested segment names that
[SegmentStatusChecker] relies on. A segment missing from
+ /// `segmentZKMetadataMap` reads back as having no ZK metadata; one missing
from `segmentZNodeMTimesMs` reads back
+ /// with no znode stat, which makes the checker fall back to the metadata's
creation time.
+ private void mockSegmentsZKMetadata(PinotHelixResourceManager
resourceManager, String tableNameWithType,
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap, Map<String, Long>
segmentZNodeMTimesMs) {
+ when(resourceManager.getSegmentsZKMetadata(eq(tableNameWithType), any(),
any())).thenAnswer(
+ invocation -> {
+ List<String> segmentNames = invocation.getArgument(1);
+ List<Stat> stats = invocation.getArgument(2);
+ List<SegmentZKMetadata> segmentsZKMetadata = new
ArrayList<>(segmentNames.size());
+ for (String segmentName : segmentNames) {
+ segmentsZKMetadata.add(segmentZKMetadataMap.get(segmentName));
+ if (stats != null) {
+ Long mTimeMs = segmentZNodeMTimesMs.get(segmentName);
+ stats.add(mTimeMs != null ? mockStatWithMTime(mTimeMs) : null);
+ }
+ }
+ return segmentsZKMetadata;
+ });
+ }
+
+ /// Stubs the single batched segment read with metadata only, so that every
segment reads back without a znode stat
+ /// and the checker falls back to the metadata's creation time
+ /// (see [#mockSegmentsZKMetadata(PinotHelixResourceManager, String, Map,
Map)]).
+ private void mockSegmentsZKMetadata(PinotHelixResourceManager
resourceManager, String tableNameWithType,
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap) {
+ mockSegmentsZKMetadata(resourceManager, tableNameWithType,
segmentZKMetadataMap, Map.of());
+ }
+
+ /// Stubs the single batched segment read so that every segment of
`idealState` resolves to `segmentZKMetadata`
+ /// (see [#mockSegmentsZKMetadata(PinotHelixResourceManager, String, Map)]).
+ private void mockSegmentsZKMetadataForAllSegments(PinotHelixResourceManager
resourceManager,
+ String tableNameWithType, IdealState idealState, SegmentZKMetadata
segmentZKMetadata) {
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap = new HashMap<>();
+ for (String segmentName : idealState.getPartitionSet()) {
+ segmentZKMetadataMap.put(segmentName, segmentZKMetadata);
+ }
+ mockSegmentsZKMetadata(resourceManager, tableNameWithType,
segmentZKMetadataMap);
+ }
+
/// A pauseless COMMITTING segment whose replicas are still building (only
1/3 ONLINE in the external view) must not
/// be counted as under-replicated while it is within the grace window, so
percentOfReplicas stays at 100. Regression
/// test for the SegmentReplicasCriticallyLowForHATable false positive on
pauseless tables.
@@ -608,15 +652,15 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committingSegmentZKMetadata =
mockCommittingSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg)).thenReturn(committingSegmentZKMetadata);
+ // Just committed: znode mtime is now, within the grace window.
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME, Map.of(seg,
committingSegmentZKMetadata),
+ Map.of(seg, System.currentTimeMillis()));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
ZNRecord znRecord = new ZNRecord("0");
znRecord.setSimpleField(CommonConstants.Segment.Realtime.END_OFFSET,
"10000");
when(propertyStore.get(anyString(), any(), anyInt())).thenReturn(znRecord);
- // Just committed: znode mtime is now, within the grace window.
- when(propertyStore.getStat(anyString(),
anyInt())).thenReturn(mockStatWithMTime(System.currentTimeMillis()));
// 1h grace window; the segment was just created, so it must be skipped
and the table stays fully replicated.
runSegmentStatusChecker(resourceManager, 3600);
@@ -626,6 +670,221 @@ public class SegmentStatusCheckerTest {
ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), 0);
}
+ /// When a segment's znode stat is unavailable, the grace window falls back
to the metadata's creation time. A freshly
+ /// created CONSUMING segment (mtime == creation time) must therefore still
be graced, so a lost stat cannot turn into
+ /// a false under-replication alert.
+ @Test
+ public void
realtimeConsumingSegmentWithoutZNodeStatFallsBackToCreationTime() {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName("timeColumn")
+ .setNumReplicas(3).setStreamConfigs(getStreamConfigMap()).build();
+
+ String seg = new LLCSegmentName(RAW_TABLE_NAME, 1, 5,
System.currentTimeMillis()).getSegmentName();
+ IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
+ idealState.setPartitionState(seg, "pinot1", "CONSUMING");
+ idealState.setPartitionState(seg, "pinot2", "CONSUMING");
+ idealState.setPartitionState(seg, "pinot3", "CONSUMING");
+ idealState.setReplicas("3");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ // Just created: only 1 of 3 replicas has started consuming.
+ ExternalView externalView = new ExternalView(REALTIME_TABLE_NAME);
+ externalView.setState(seg, "pinot1", "CONSUMING");
+ externalView.setState(seg, "pinot2", "OFFLINE");
+ externalView.setState(seg, "pinot3", "OFFLINE");
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+
when(resourceManager.getHelixInstanceConfig(any())).thenReturn(newQuerableInstanceConfig("any"));
+
when(resourceManager.getTableConfig(REALTIME_TABLE_NAME)).thenReturn(tableConfig);
+
when(resourceManager.getAllTables()).thenReturn(List.of(REALTIME_TABLE_NAME));
+
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
+
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
+ // Creation time is now, and no znode stat is available -> the fallback
must keep the segment within the grace
+ // window
+ SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(System.currentTimeMillis());
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME, Map.of(seg,
consumingSegmentZKMetadata));
+
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
+ when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
+ ZNRecord znRecord = new ZNRecord("0");
+ znRecord.setSimpleField(CommonConstants.Segment.Realtime.END_OFFSET,
"10000");
+ when(propertyStore.get(anyString(), any(), anyInt())).thenReturn(znRecord);
+
+ // 1h grace window; the segment was just created, so it must be skipped
and the table stays fully replicated.
+ runSegmentStatusChecker(resourceManager, 3600);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.PERCENT_OF_REPLICAS), 100);
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
REALTIME_TABLE_NAME,
+ ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), 0);
+ }
+
+ @DataProvider(name = "segmentMetadataBatchSizes")
+ public Object[][] segmentMetadataBatchSizes() {
+ // Batch size (null leaves the production default, under which the whole
table fits in one batch) and the number of
+ // reads the 7 segments must then take. 3 does not divide 7 evenly, so the
last batch is a partial one.
+ return new Object[][]{
+ {null, 1},
+ {3, 3}
+ };
+ }
+
+ /// The metadata and znode stats come back from batched reads, so each
segment must be matched to its own metadata
+ /// (by name) and its own stat (by position) rather than to whichever entry
happens to sit at its index, whether the
+ /// table is read in one batch or in several. Every segment is
under-replicated and carries a distinct size, exactly
+ /// one segment was pushed recently enough to be graced, and the last
segment is the only one with 4 replicas so it
+ /// alone determines PERCENT_OF_REPLICAS. Together the gauges pin that every
segment was examined under its own name,
+ /// that the grace window applied to exactly the segment whose znode stat is
recent, and that the sizes accumulate.
+ /// One segment has no ZK metadata, which shifts the alignment if the
pairing gets it wrong.
+ @Test(dataProvider = "segmentMetadataBatchSizes")
+ public void segmentsStayAlignedWithTheirBatchedMetadata(Integer
segmentMetadataBatchSize, int expectedNumBatches) {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(2).build();
+
+ int numSegments = 7;
+ // Distinct positions, none of them the last segment, which carries its
own marker below. With a batch size of 3 the
+ // segment without ZK metadata also starts a batch, so a boundary that
shifted the pairing drops the wrong segment.
+ int segmentWithoutZKMetadata = 3;
+ int recentlyPushedSegment = 1;
+ // The last segment is the marker that pins name-to-metadata pairing: it
is the only one whose replica ratio is 1/4
+ // rather than 1/2, so PERCENT_OF_REPLICAS drops to 25 only if this exact
segment was examined.
+ int lowReplicaSegment = numSegments - 1;
+ long oldPushTimeMs = 11111L;
+
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, SegmentZKMetadata> segmentZKMetadataMap = new HashMap<>();
+ Map<String, Long> segmentZNodeMTimesMs = new HashMap<>();
+ long expectedTableCompressedSize = 0;
+ for (int i = 0; i < numSegments; i++) {
+ String segment = "myTable_" + i;
+ int numReplicas = i == lowReplicaSegment ? 4 : 2;
+ // Every segment is under-replicated, so any segment that is examined
and not graced must be counted
+ for (int replica = 1; replica <= numReplicas; replica++) {
+ idealState.setPartitionState(segment, "pinot" + replica, "ONLINE");
+ externalView.setState(segment, "pinot" + replica, replica == 1 ?
"ONLINE" : "OFFLINE");
+ }
+ if (i == segmentWithoutZKMetadata) {
+ continue;
+ }
+ // Distinct size per segment so that the total pins which metadata was
attributed to which segment
+ long sizeInBytes = 1000L + i;
+ segmentZKMetadataMap.put(segment,
mockPushedSegmentZKMetadata(sizeInBytes, oldPushTimeMs));
+ segmentZNodeMTimesMs.put(segment, i == recentlyPushedSegment ?
System.currentTimeMillis() : oldPushTimeMs);
+ expectedTableCompressedSize += sizeInBytes;
+ }
+ idealState.setReplicas("2");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+
when(resourceManager.getHelixInstanceConfig(any())).thenReturn(newQuerableInstanceConfig("any"));
+
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
+
when(resourceManager.getTableConfig(OFFLINE_TABLE_NAME)).thenReturn(tableConfig);
+
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
+
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
+ mockSegmentsZKMetadata(resourceManager, OFFLINE_TABLE_NAME,
segmentZKMetadataMap, segmentZNodeMTimesMs);
+
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
+ when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
+
+ // 10min grace window, so only the recently pushed segment is skipped
+ SegmentStatusChecker segmentStatusChecker =
+ buildSegmentStatusChecker(resourceManager, 600,
mock(TableSizeReader.class));
+ if (segmentMetadataBatchSize != null) {
+ segmentStatusChecker._segmentMetadataBatchSize =
segmentMetadataBatchSize;
+ }
+ runSegmentStatusChecker(segmentStatusChecker);
+
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SEGMENT_COUNT), numSegments);
+ // Every segment except the one without ZK metadata and the graced one
must be counted. A segment paired with the
+ // wrong metadata or the wrong znode stat drops out of, or into, this
count.
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS), numSegments - 2);
+ // Only the last segment has a 1-of-4 ratio, so this is 25 only if that
segment was examined under its own name
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.PERCENT_OF_REPLICAS), 25);
+ // The sizes accumulate over exactly the segments that have metadata
(including the graced one, whose size is
+ // counted before the grace check)
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.TABLE_COMPRESSED_SIZE), expectedTableCompressedSize);
+
+ // The reads must be batched, metadata and znode stats together: one
request per batch and no more, otherwise the
+ // per-segment reads crept back in some form
+ ArgumentCaptor<List<String>> segmentNamesCaptor =
ArgumentCaptor.forClass(List.class);
+ verify(resourceManager,
times(expectedNumBatches)).getSegmentsZKMetadata(eq(OFFLINE_TABLE_NAME),
+ segmentNamesCaptor.capture(), any());
+ verify(propertyStore, never()).getStats(any(), anyInt());
+ // Every segment is requested exactly once, in batches of at most the
batch size
+ List<String> requestedSegments = new ArrayList<>();
+ for (List<String> batch : segmentNamesCaptor.getAllValues()) {
+ assertTrue(segmentMetadataBatchSize == null || batch.size() <=
segmentMetadataBatchSize,
+ "batch of " + batch.size() + " segments");
+ requestedSegments.addAll(batch);
+ }
+ assertEquals(requestedSegments.size(), numSegments);
+ assertEquals(new HashSet<>(requestedSegments),
idealState.getPartitionSet());
+ }
+
+ /// When not a single segment's ZK metadata can be read the table's gauges
must be left alone rather than reset to
+ /// all-green values, because an all-green gauge silences the alerts that a
stale one would still fire. Regression
+ /// test for a whole-table ZK read failure being reported as a perfectly
healthy table. This supersedes the former
+ /// `noSegmentZKMetadataTest`, which expected the all-green gauges for the
same scenario.
+ @Test
+ public void tableWithoutAnyReadableSegmentZKMetadataKeepsItsGauges() {
+ TableConfig tableConfig =
+ new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(2).build();
+
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ for (int i = 0; i < 3; i++) {
+ String segment = "myTable_" + i;
+ idealState.setPartitionState(segment, "pinot1", "ONLINE");
+ idealState.setPartitionState(segment, "pinot2", "ONLINE");
+ // Every segment is under-replicated, so a metric update that went ahead
would be visibly wrong
+ externalView.setState(segment, "pinot1", "ONLINE");
+ externalView.setState(segment, "pinot2", "OFFLINE");
+ }
+ idealState.setReplicas("2");
+ idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
+
+ PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
+
when(resourceManager.getHelixInstanceConfig(any())).thenReturn(newQuerableInstanceConfig("any"));
+
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
+
when(resourceManager.getTableConfig(OFFLINE_TABLE_NAME)).thenReturn(tableConfig);
+
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
+
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
+ // No segment resolves to metadata, as if every znode read had failed
+ mockSegmentsZKMetadata(resourceManager, OFFLINE_TABLE_NAME, Map.of());
+
+ ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
+ when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
+
+ // Sentinels standing in for what a previous, successful cycle had
published. None of them is a value this table
+ // could legitimately produce, so any of them being overwritten means the
checker went ahead on unreadable metadata.
+ List<ControllerGauge> segmentHealthGauges =
+ List.of(ControllerGauge.PERCENT_OF_REPLICAS,
ControllerGauge.SEGMENTS_WITH_LESS_REPLICAS,
+ ControllerGauge.PERCENT_SEGMENTS_AVAILABLE,
ControllerGauge.SEGMENTS_IN_ERROR_STATE,
+ ControllerGauge.TABLE_COMPRESSED_SIZE);
+ for (ControllerGauge gauge : segmentHealthGauges) {
+ _controllerMetrics.setValueOfTableGauge(OFFLINE_TABLE_NAME, gauge, -1);
+ }
+
+ runSegmentStatusChecker(resourceManager, 600);
+
+ // SEGMENT_COUNT is published from the ideal state before the read, so it
is still updated
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME,
+ ControllerGauge.SEGMENT_COUNT), 3);
+ for (ControllerGauge gauge : segmentHealthGauges) {
+ assertEquals(MetricValueUtils.getTableGaugeValue(_controllerMetrics,
OFFLINE_TABLE_NAME, gauge), -1,
+ gauge.getGaugeName());
+ }
+
+ // The metrics are shared by every test in this class, so do not leave the
sentinels behind
+ for (ControllerGauge gauge : segmentHealthGauges) {
+ _controllerMetrics.removeTableGauge(OFFLINE_TABLE_NAME, gauge);
+ }
+ }
+
/// A COMMITTING segment that has been under-replicated for longer than the
grace window is a genuinely stuck commit
/// and must still be flagged (percentOfReplicas drops), so the grace does
not mask real problems.
@Test
@@ -654,16 +913,15 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata committingSegmentZKMetadata =
mockCommittingSegmentZKMetadata();
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
seg)).thenReturn(committingSegmentZKMetadata);
+ // Committed 2h ago (znode mtime), still under-replicated -> a stuck
commit, must not be graced.
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME, Map.of(seg,
committingSegmentZKMetadata),
+ Map.of(seg, System.currentTimeMillis() - 7200000L));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
ZNRecord znRecord = new ZNRecord("0");
znRecord.setSimpleField(CommonConstants.Segment.Realtime.END_OFFSET,
"10000");
when(propertyStore.get(anyString(), any(), anyInt())).thenReturn(znRecord);
- // Committed 2h ago (znode mtime), still under-replicated -> a stuck
commit, must not be graced.
- when(propertyStore.getStat(anyString(), anyInt()))
- .thenReturn(mockStatWithMTime(System.currentTimeMillis() - 7200000L));
// 1h grace window; the segment is 2h old and still 1/3 replicas up, so it
must be flagged (33%).
runSegmentStatusChecker(resourceManager, 3600);
@@ -699,7 +957,7 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata segmentZKMetadata = mockPushedSegmentZKMetadata(1234,
11111L);
- when(resourceManager.getSegmentZKMetadata(eq(OFFLINE_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, OFFLINE_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -724,7 +982,7 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
SegmentZKMetadata segmentZKMetadata = mockPushedSegmentZKMetadata(1234,
11111L);
- when(resourceManager.getSegmentZKMetadata(eq(OFFLINE_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, OFFLINE_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -790,19 +1048,15 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata segmentZKMetadata01 = mockPushedSegmentZKMetadata(1234,
11111L);
- when(resourceManager.getSegmentZKMetadata(OFFLINE_TABLE_NAME,
"myTable_0")).thenReturn(segmentZKMetadata01);
- when(resourceManager.getSegmentZKMetadata(OFFLINE_TABLE_NAME,
"myTable_1")).thenReturn(segmentZKMetadata01);
SegmentZKMetadata segmentZKMetadata2 = mockPushedSegmentZKMetadata(1234,
System.currentTimeMillis());
- when(resourceManager.getSegmentZKMetadata(OFFLINE_TABLE_NAME,
"myTable_2")).thenReturn(segmentZKMetadata2);
+ // myTable_2 was just pushed (znode mtime is now) so it is within the
grace window and skipped; the others were
+ // pushed long ago.
+ mockSegmentsZKMetadata(resourceManager, OFFLINE_TABLE_NAME,
+ Map.of("myTable_0", segmentZKMetadata01, "myTable_1",
segmentZKMetadata01, "myTable_2", segmentZKMetadata2),
+ Map.of("myTable_0", 11111L, "myTable_1", 11111L, "myTable_2",
System.currentTimeMillis()));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
- // myTable_2 was just pushed (znode mtime is now) so it is within the
grace window and skipped; the others were
- // pushed long ago.
- when(propertyStore.getStat(anyString(), anyInt())).thenAnswer(inv -> {
- String path = inv.getArgument(0);
- return mockStatWithMTime(path.endsWith("myTable_2") ?
System.currentTimeMillis() : 11111L);
- });
runSegmentStatusChecker(resourceManager, 600);
verifyControllerMetrics(OFFLINE_TABLE_NAME, 0, 3, 3, 2, 100, 0, 100, 0,
3702);
@@ -820,14 +1074,14 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getAllTables()).thenReturn(List.of(REALTIME_TABLE_NAME));
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
SegmentZKMetadata updatedSegmentZKMetadata =
mockPushedSegmentZKMetadata(1234, System.currentTimeMillis());
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
"myTable_0")).thenReturn(updatedSegmentZKMetadata);
SegmentZKMetadata consumingSegmentZKMetadata =
mockConsumingSegmentZKMetadata(System.currentTimeMillis());
- when(resourceManager.getSegmentZKMetadata(REALTIME_TABLE_NAME,
"myTable_1")).thenReturn(consumingSegmentZKMetadata);
+ // Both segments were just updated/created (znode mtime is now), so they
are within the grace window and skipped.
+ mockSegmentsZKMetadata(resourceManager, REALTIME_TABLE_NAME,
+ Map.of("myTable_0", updatedSegmentZKMetadata, "myTable_1",
consumingSegmentZKMetadata),
+ Map.of("myTable_0", System.currentTimeMillis(), "myTable_1",
System.currentTimeMillis()));
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
- // Both segments were just updated/created (znode mtime is now), so they
are within the grace window and skipped.
- when(propertyStore.getStat(anyString(),
anyInt())).thenReturn(mockStatWithMTime(System.currentTimeMillis()));
runSegmentStatusChecker(resourceManager, 600);
verifyControllerMetrics(REALTIME_TABLE_NAME, 0, 2, 2, 1, 100, 0, 100, 0,
1234);
@@ -847,7 +1101,7 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(REALTIME_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(REALTIME_TABLE_NAME)).thenReturn(null);
SegmentZKMetadata segmentZKMetadata =
mockConsumingSegmentZKMetadata(11111L);
- when(resourceManager.getSegmentZKMetadata(eq(REALTIME_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, REALTIME_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -856,24 +1110,6 @@ public class SegmentStatusCheckerTest {
verifyControllerMetrics(REALTIME_TABLE_NAME, 0, 1, 1, 1, 100, 0, 100, 0,
0);
}
- @Test
- public void noSegmentZKMetadataTest() {
- IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
- idealState.setPartitionState("myTable_0", "pinot1", "ONLINE");
- idealState.setReplicas("1");
- idealState.setRebalanceMode(IdealState.RebalanceMode.CUSTOMIZED);
-
- PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
-
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
-
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
-
- ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
- when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
-
- runSegmentStatusChecker(resourceManager, 0);
- verifyControllerMetrics(OFFLINE_TABLE_NAME, 0, 1, 1, 1, 100, 0, 100, 0, 0);
- }
-
@Test
public void disabledTableTest()
throws Exception {
@@ -946,7 +1182,7 @@ public class SegmentStatusCheckerTest {
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
when(resourceManager.getTableExternalView(OFFLINE_TABLE_NAME)).thenReturn(externalView);
SegmentZKMetadata segmentZKMetadata = mockPushedSegmentZKMetadata(1234,
11111L);
- when(resourceManager.getSegmentZKMetadata(eq(OFFLINE_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, OFFLINE_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -1157,7 +1393,7 @@ public class SegmentStatusCheckerTest {
PinotHelixResourceManager resourceManager =
mock(PinotHelixResourceManager.class);
when(resourceManager.getAllTables()).thenReturn(List.of(OFFLINE_TABLE_NAME));
when(resourceManager.getTableIdealState(OFFLINE_TABLE_NAME)).thenReturn(idealState);
- when(resourceManager.getSegmentZKMetadata(eq(OFFLINE_TABLE_NAME),
anyString())).thenReturn(segmentZKMetadata);
+ mockSegmentsZKMetadataForAllSegments(resourceManager, OFFLINE_TABLE_NAME,
idealState, segmentZKMetadata);
ZkHelixPropertyStore<ZNRecord> propertyStore =
mock(ZkHelixPropertyStore.class);
when(resourceManager.getPropertyStore()).thenReturn(propertyStore);
@@ -1336,11 +1572,15 @@ public class SegmentStatusCheckerTest {
private SegmentStatusChecker
buildSegmentStatusChecker(PinotHelixResourceManager resourceManager,
int waitForPushTimeInSeconds) {
+ return buildSegmentStatusChecker(resourceManager,
waitForPushTimeInSeconds, mock(TableSizeReader.class));
+ }
+
+ private SegmentStatusChecker
buildSegmentStatusChecker(PinotHelixResourceManager resourceManager,
+ int waitForPushTimeInSeconds, TableSizeReader tableSizeReader) {
LeadControllerManager leadControllerManager =
mock(LeadControllerManager.class);
when(leadControllerManager.isLeaderForTable(anyString())).thenReturn(true);
ControllerConf controllerConf = mock(ControllerConf.class);
when(controllerConf.getStatusCheckerWaitForPushTimeInSeconds()).thenReturn(waitForPushTimeInSeconds);
- TableSizeReader tableSizeReader = mock(TableSizeReader.class);
return new SegmentStatusChecker(resourceManager, leadControllerManager,
controllerConf, _controllerMetrics,
tableSizeReader);
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
index c04c521dd1c..64a8ad22274 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManagerStatelessTest.java
@@ -91,6 +91,7 @@ import org.apache.pinot.spi.utils.CommonConstants.Server;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.util.TestUtils;
+import org.apache.zookeeper.data.Stat;
import org.joda.time.DateTimeZone;
import org.joda.time.format.DateTimeFormatter;
import org.joda.time.format.DateTimeFormatterBuilder;
@@ -588,6 +589,51 @@ public class PinotHelixResourceManagerStatelessTest
extends ControllerTest {
}
}
+ /// Pins the contract
[org.apache.pinot.controller.helix.SegmentStatusChecker] depends on: the
batched read returns
+ /// the metadata and the znode [Stat] index-aligned with the requested
segment names, with `null` in both for a
+ /// segment that has no ZK metadata.
+ @Test
+ public void testRetrieveSegmentsZKMetadataBatched() {
+ long beforeMs = System.currentTimeMillis();
+ List<String> segmentNames = List.of("testSegment0", "testSegment1",
"testSegment2");
+ // Write ZK metadata for the first and last segment only, leaving a gap in
the middle
+ for (String segmentName : List.of("testSegment0", "testSegment2")) {
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segmentName);
+ segmentZKMetadata.setSizeInBytes(segmentName.hashCode());
+ ZKMetadataProvider.setSegmentZKMetadata(_propertyStore,
OFFLINE_TABLE_NAME, segmentZKMetadata);
+ }
+
+ try {
+ List<Stat> stats = new ArrayList<>();
+ List<SegmentZKMetadata> segmentsZKMetadata =
+ _helixResourceManager.getSegmentsZKMetadata(OFFLINE_TABLE_NAME,
segmentNames, stats);
+
+ // Both lists must line up with the requested names, so that the gap
does not shift the entries after it
+ assertEquals(segmentsZKMetadata.size(), 3);
+ assertEquals(stats.size(), 3);
+ for (int i : new int[]{0, 2}) {
+ String segmentName = segmentNames.get(i);
+ assertNotNull(segmentsZKMetadata.get(i), segmentName);
+ assertEquals(segmentsZKMetadata.get(i).getSegmentName(), segmentName);
+ assertEquals(segmentsZKMetadata.get(i).getSizeInBytes(),
segmentName.hashCode());
+ // A real znode mtime, not a default or the metadata's own creation
time
+ assertNotNull(stats.get(i), segmentName);
+ assertTrue(stats.get(i).getMtime() >= beforeMs,
+ segmentName + " mtime: " + stats.get(i).getMtime() + " < " +
beforeMs);
+ }
+ assertNull(segmentsZKMetadata.get(1));
+ assertNull(stats.get(1));
+
+ // The stats are optional
+ assertEquals(
+ _helixResourceManager.getSegmentsZKMetadata(OFFLINE_TABLE_NAME,
segmentNames, null).size(), 3);
+ } finally {
+ for (String segmentName : List.of("testSegment0", "testSegment2")) {
+ ZKMetadataProvider.removeSegmentZKMetadata(_propertyStore,
OFFLINE_TABLE_NAME, segmentName);
+ }
+ }
+ }
+
@Test
public void testUpdateSchemaDateTime() {
String segmentName = "testSegment";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
