This is an automated email from the ASF dual-hosted git repository.
gortiz 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 129fa937f34 Make broker routing components update incrementally on
assignment changes (#19783)
129fa937f34 is described below
commit 129fa937f34492c07b734dcd51d9a9b57f718e44
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Thu Oct 15 10:20:03 2026 +0200
Make broker routing components update incrementally on assignment changes
(#19783)
Part of #19782.
---
.../SegmentPartitionMetadataManager.java | 117 +++++++++++++++---
.../segmentpartition/SegmentPartitionUtils.java | 66 +++++++++-
.../routing/timeboundary/TimeBoundaryManager.java | 77 +++++++++---
.../SegmentPartitionMetadataManagerTest.java | 133 +++++++++++++++++++++
.../SegmentPartitionUtilsTest.java | 108 +++++++++++++++++
.../timeboundary/TimeBoundaryManagerTest.java | 121 +++++++++++++++++++
6 files changed, 586 insertions(+), 36 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
index 1cf6ed6a465..cf41e32ee1d 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManager.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.broker.routing.segmentpartition;
+import com.google.common.annotations.VisibleForTesting;
+import java.time.Clock;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -63,10 +65,18 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
private final String _partitionFunctionName;
private final int _numPartitions;
private final long _newSegmentExpirationMs;
+ private final Clock _clock;
// cache-able content, only follow changes if onlineSegments list (of
ideal-state) is changed.
private final Map<String, SegmentInfo> _segmentInfoMap = new HashMap<>();
+ // Among the segments treated as new by the last
computeTablePartitionReplicatedServersInfo(), the last time at which
+ // the first of them to expire is still new, or Long.MAX_VALUE when there is
no new segment.
+ // Newness is re-evaluated against the clock only when the partition info is
recomputed, so an assignment change that
+ // does not otherwise change anything must still recompute once this time
has passed.
+ // Writer-private: only read and written by the rebuild thread.
+ private long _nextNewSegmentExpirationTimeMs = Long.MAX_VALUE;
+
// computed value based on status change.
// NOTE: Volatile because they are written while the table's routing entry
is built or updated, and read without any
// lock by unrelated threads (e.g. query planner threads). The writers are
serialized by BaseBrokerRoutingManager,
@@ -78,22 +88,31 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
public SegmentPartitionMetadataManager(String tableNameWithType, String
partitionColumn, String partitionFunctionName,
int numPartitions, long newSegmentExpirationMs) {
+ this(tableNameWithType, partitionColumn, partitionFunctionName,
numPartitions, newSegmentExpirationMs,
+ Clock.systemUTC());
+ }
+
+ @VisibleForTesting
+ SegmentPartitionMetadataManager(String tableNameWithType, String
partitionColumn, String partitionFunctionName,
+ int numPartitions, long newSegmentExpirationMs, Clock clock) {
_tableNameWithType = tableNameWithType;
_partitionColumn = partitionColumn;
_partitionFunctionName = partitionFunctionName;
_numPartitions = numPartitions;
_newSegmentExpirationMs = newSegmentExpirationMs;
+ _clock = clock;
}
@Override
public void init(IdealState idealState, ExternalView externalView,
List<String> onlineSegments,
List<ZNRecord> znRecords) {
+ Map<List<String>, List<String>> onlineServersCache = new HashMap<>();
int numSegments = onlineSegments.size();
for (int i = 0; i < numSegments; i++) {
String segment = onlineSegments.get(i);
ZNRecord znRecord = znRecords.get(i);
SegmentInfo segmentInfo = new SegmentInfo(getPartitionId(segment,
znRecord), getCreationTimeMs(znRecord),
- getOnlineServers(externalView, segment));
+ getOnlineServers(externalView.getStateMap(segment),
onlineServersCache));
_segmentInfoMap.put(segment, segmentInfo);
}
computeAllTablePartitionInfo();
@@ -129,49 +148,104 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
return SegmentUtils.getSegmentCreationTimeMs(new
SegmentZKMetadata(znRecord));
}
- private static List<String> getOnlineServers(ExternalView externalView,
String segment) {
- Map<String, String> instanceStateMap = externalView.getStateMap(segment);
+ /// Returns the immutable list of servers hosting the segment in ONLINE or
CONSUMING state.
+ ///
+ /// Equal lists are shared through `onlineServersCache`, so that the many
segments hosted on the same servers (e.g.
+ /// with replica-group assignment) keep a single list instead of one per
segment.
+ private static List<String> getOnlineServers(@Nullable Map<String, String>
instanceStateMap,
+ Map<List<String>, List<String>> onlineServersCache) {
if (instanceStateMap == null) {
return List.of();
}
List<String> onlineServers = new ArrayList<>(instanceStateMap.size());
for (Map.Entry<String, String> entry : instanceStateMap.entrySet()) {
- String instanceState = entry.getValue();
- if (instanceState.equals(SegmentStateModel.ONLINE) ||
instanceState.equals(SegmentStateModel.CONSUMING)) {
+ if (isOnline(entry.getValue())) {
onlineServers.add(entry.getKey());
}
}
- return onlineServers;
+ if (onlineServers.isEmpty()) {
+ return List.of();
+ }
+ return onlineServersCache.computeIfAbsent(onlineServers,
Collections::unmodifiableList);
+ }
+
+ private static boolean isOnline(String instanceState) {
+ return instanceState.equals(SegmentStateModel.ONLINE) ||
instanceState.equals(SegmentStateModel.CONSUMING);
+ }
+
+ /// Returns whether `onlineServers` holds exactly the servers in ONLINE or
CONSUMING state in `instanceStateMap`.
+ /// Does not allocate, so that the common case of an unchanged segment costs
no garbage.
+ private static boolean hasSameOnlineServers(List<String> onlineServers,
+ @Nullable Map<String, String> instanceStateMap) {
+ if (instanceStateMap == null) {
+ return onlineServers.isEmpty();
+ }
+ int numOnlineServers = 0;
+ for (Map.Entry<String, String> entry : instanceStateMap.entrySet()) {
+ if (isOnline(entry.getValue())) {
+ // NOTE: The list holds the keys of a map, so it has no duplicates and
a size check is enough after this.
+ if (!onlineServers.contains(entry.getKey())) {
+ return false;
+ }
+ numOnlineServers++;
+ }
+ }
+ return numOnlineServers == onlineServers.size();
}
@Override
public synchronized void onAssignmentChange(IdealState idealState,
ExternalView externalView,
Set<String> onlineSegments, List<String> pulledSegments, List<ZNRecord>
znRecords) {
+ // Both published objects are recomputed only when their inputs change:
TablePartitionInfo depends on the segments
+ // and their partition ids, TablePartitionReplicatedServersInfo also
depends on the online servers of each segment
+ // and on the clock (see _nextNewSegmentExpirationTimeMs).
+ boolean partitionsChanged = false;
+ boolean onlineServersChanged = false;
+ Map<List<String>, List<String>> onlineServersCache = new HashMap<>();
// Update segment partition id for the pulled segments
+ // NOTE: Segments without ZK metadata are pulled again on every change, so
compare with the existing info.
int numSegments = pulledSegments.size();
for (int i = 0; i < numSegments; i++) {
String segment = pulledSegments.get(i);
ZNRecord znRecord = znRecords.get(i);
- SegmentInfo segmentInfo = new SegmentInfo(getPartitionId(segment,
znRecord), getCreationTimeMs(znRecord),
- getOnlineServers(externalView, segment));
- _segmentInfoMap.put(segment, segmentInfo);
+ int partitionId = getPartitionId(segment, znRecord);
+ long creationTimeMs = getCreationTimeMs(znRecord);
+ SegmentInfo segmentInfo = _segmentInfoMap.get(segment);
+ if (segmentInfo == null) {
+ _segmentInfoMap.put(segment, new SegmentInfo(partitionId,
creationTimeMs,
+ getOnlineServers(externalView.getStateMap(segment),
onlineServersCache)));
+ partitionsChanged = true;
+ } else if (segmentInfo._partitionId != partitionId ||
segmentInfo._creationTimeMs != creationTimeMs) {
+ segmentInfo._partitionId = partitionId;
+ segmentInfo._creationTimeMs = creationTimeMs;
+ partitionsChanged = true;
+ }
}
// Update online servers for all online segments
for (String segment : onlineSegments) {
+ Map<String, String> instanceStateMap = externalView.getStateMap(segment);
SegmentInfo segmentInfo = _segmentInfoMap.get(segment);
if (segmentInfo == null) {
// NOTE: This should not happen, but we still handle it gracefully by
adding an invalid SegmentInfo
LOGGER.error("Failed to find segment info for segment: {} in table: {}
while handling assignment change",
segment, _tableNameWithType);
- segmentInfo =
- new SegmentInfo(INVALID_PARTITION_ID, INVALID_CREATION_TIME_MS,
getOnlineServers(externalView, segment));
+ segmentInfo = new SegmentInfo(INVALID_PARTITION_ID,
INVALID_CREATION_TIME_MS,
+ getOnlineServers(instanceStateMap, onlineServersCache));
_segmentInfoMap.put(segment, segmentInfo);
- } else {
- segmentInfo._onlineServers = getOnlineServers(externalView, segment);
+ partitionsChanged = true;
+ } else if (!hasSameOnlineServers(segmentInfo._onlineServers,
instanceStateMap)) {
+ segmentInfo._onlineServers = getOnlineServers(instanceStateMap,
onlineServersCache);
+ onlineServersChanged = true;
}
}
- _segmentInfoMap.keySet().retainAll(onlineSegments);
- computeAllTablePartitionInfo();
+ if (_segmentInfoMap.keySet().retainAll(onlineSegments)) {
+ partitionsChanged = true;
+ }
+ if (partitionsChanged) {
+ computeAllTablePartitionInfo();
+ } else if (onlineServersChanged || _clock.millis() >
_nextNewSegmentExpirationTimeMs) {
+ computeTablePartitionReplicatedServersInfo();
+ }
}
@Override
@@ -211,7 +285,8 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
List<Pair<String, Integer>> unavailableSegments = new ArrayList<>();
List<Triple<String, Integer, Integer>>
segmentsReducingFullyReplicatedServers = new ArrayList<>();
List<Map.Entry<String, SegmentInfo>> newSegmentInfoEntries = new
ArrayList<>();
- long currentTimeMs = System.currentTimeMillis();
+ long currentTimeMs = _clock.millis();
+ long nextNewSegmentExpirationTimeMs = Long.MAX_VALUE;
for (Map.Entry<String, SegmentInfo> entry : _segmentInfoMap.entrySet()) {
String segment = entry.getKey();
SegmentInfo segmentInfo = entry.getValue();
@@ -223,6 +298,8 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
// Process new segments in the end
if (InstanceSelector.isNewSegment(segmentInfo._creationTimeMs,
currentTimeMs, _newSegmentExpirationMs)) {
newSegmentInfoEntries.add(entry);
+ nextNewSegmentExpirationTimeMs =
+ Math.min(nextNewSegmentExpirationTimeMs,
getNewSegmentExpirationTimeMs(segmentInfo._creationTimeMs));
continue;
}
List<String> onlineServers = segmentInfo._onlineServers;
@@ -333,11 +410,19 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
_tableNameWithType);
}
}
+ _nextNewSegmentExpirationTimeMs = nextNewSegmentExpirationTimeMs;
_tablePartitionReplicatedServersInfo =
new TablePartitionReplicatedServersInfo(_tableNameWithType,
_partitionColumn, _partitionFunctionName,
_numPartitions, partitionInfoMap, segmentsWithInvalidPartition,
partitionsWithOnlyDeferredSegments);
}
+ /// Returns the time after which a segment created at `creationTimeMs` is no
longer new, i.e. the last time at which
+ /// [InstanceSelector#isNewSegment] still returns `true` for it. Saturates
instead of overflowing.
+ private long getNewSegmentExpirationTimeMs(long creationTimeMs) {
+ return creationTimeMs > Long.MAX_VALUE - _newSegmentExpirationMs ?
Long.MAX_VALUE
+ : creationTimeMs + _newSegmentExpirationMs;
+ }
+
private void computeTablePartitionInfo() {
List<List<String>> segmentsByPartition = new ArrayList<>();
for (int i = 0; i < _numPartitions; i++) {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtils.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtils.java
index 5821b498815..9a8e359f5a3 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtils.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtils.java
@@ -18,12 +18,16 @@
*/
package org.apache.pinot.broker.routing.segmentpartition;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import it.unimi.dsi.fastutil.ints.IntSets;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Nullable;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
import org.apache.pinot.spi.utils.CommonConstants;
@@ -40,6 +44,26 @@ public class SegmentPartitionUtils {
private static final Logger LOGGER =
LoggerFactory.getLogger(SegmentPartitionUtils.class);
+ // Shared partition functions, keyed by the metadata they are built from.
Every segment of a table usually carries the
+ // same partition function metadata, so sharing one instance avoids creating
one per segment on every routing
+ // rebuild. Sharing is safe because PartitionFunction implementations must
be stateless and thread-safe (see the
+ // PartitionFunction contract). The cache stops growing at
MAX_CACHED_PARTITION_FUNCTIONS distinct entries, after
+ // which new functions are created without being cached.
+ private static final int MAX_CACHED_PARTITION_FUNCTIONS = 1024;
+ private static final Map<PartitionFunctionKey, PartitionFunction>
PARTITION_FUNCTION_CACHE =
+ new ConcurrentHashMap<>();
+
+ // Shared immutable one-element partition sets. Most segments hold a single
partition, so sharing them avoids one set
+ // per segment.
+ private static final int NUM_CACHED_SINGLE_PARTITION_SETS = 1024;
+ private static final IntSet[] SINGLE_PARTITION_SETS = new
IntSet[NUM_CACHED_SINGLE_PARTITION_SETS];
+
+ static {
+ for (int i = 0; i < NUM_CACHED_SINGLE_PARTITION_SETS; i++) {
+ SINGLE_PARTITION_SETS[i] = IntSets.singleton(i);
+ }
+ }
+
/// Returns the partition info for a given segment with single partition
column.
///
/// NOTE: Returns `null` when the ZNRecord is missing (could be transient
Helix issue). Returns
@@ -76,9 +100,42 @@ public class SegmentPartitionUtils {
return INVALID_PARTITION_INFO;
}
- return new SegmentPartitionInfo(partitionColumn,
- PartitionFunctionFactory.getPartitionFunction(columnPartitionMetadata),
- columnPartitionMetadata.getPartitions());
+ return new SegmentPartitionInfo(partitionColumn,
getPartitionFunction(columnPartitionMetadata),
+ getPartitions(columnPartitionMetadata));
+ }
+
+ /// Returns a partition function for the given metadata, shared with the
other segments with equal metadata.
+ private static PartitionFunction
getPartitionFunction(ColumnPartitionMetadata columnPartitionMetadata) {
+ PartitionFunctionKey key = new
PartitionFunctionKey(columnPartitionMetadata.getFunctionName(),
+ columnPartitionMetadata.getNumPartitions(),
columnPartitionMetadata.getFunctionConfig());
+ PartitionFunction partitionFunction = PARTITION_FUNCTION_CACHE.get(key);
+ if (partitionFunction != null) {
+ return partitionFunction;
+ }
+ partitionFunction =
PartitionFunctionFactory.getPartitionFunction(columnPartitionMetadata);
+ if (PARTITION_FUNCTION_CACHE.size() < MAX_CACHED_PARTITION_FUNCTIONS) {
+ PartitionFunction existing = PARTITION_FUNCTION_CACHE.putIfAbsent(key,
partitionFunction);
+ if (existing != null) {
+ return existing;
+ }
+ }
+ return partitionFunction;
+ }
+
+ /// Returns the partitions of the given metadata, replacing a one-element
set with a shared immutable one.
+ private static Set<Integer> getPartitions(ColumnPartitionMetadata
columnPartitionMetadata) {
+ Set<Integer> partitions = columnPartitionMetadata.getPartitions();
+ if (partitions.size() == 1) {
+ int partition = partitions.iterator().next();
+ if (partition >= 0 && partition < NUM_CACHED_SINGLE_PARTITION_SETS) {
+ return SINGLE_PARTITION_SETS[partition];
+ }
+ }
+ return partitions;
+ }
+
+ private record PartitionFunctionKey(String functionName, int numPartitions,
+ @Nullable Map<String, String>
functionConfig) {
}
/// Returns a map from partition column name to partition info for a given
segment with multiple partition columns.
@@ -119,8 +176,7 @@ public class SegmentPartitionUtils {
continue;
}
SegmentPartitionInfo segmentPartitionInfo = new
SegmentPartitionInfo(partitionColumn,
-
PartitionFunctionFactory.getPartitionFunction(columnPartitionMetadata),
- columnPartitionMetadata.getPartitions());
+ getPartitionFunction(columnPartitionMetadata),
getPartitions(columnPartitionMetadata));
columnSegmentPartitionInfoMap.put(partitionColumn, segmentPartitionInfo);
}
if (columnSegmentPartitionInfoMap.size() == 1) {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
index 18256275af9..81396eab437 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManager.java
@@ -21,6 +21,7 @@ package org.apache.pinot.broker.routing.timeboundary;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.HashMap;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -63,6 +64,9 @@ public class TimeBoundaryManager {
private final DateTimeFormatSpec _timeFormatSpec;
private final long _timeOffsetMs;
private final Map<String, Long> _endTimeMsMap = new HashMap<>();
+ // Max value in _endTimeMsMap, or INVALID_TIME_MS when it is empty. Kept in
sync with the map so that an assignment
+ // change does not scan all the segments.
+ private long _maxEndTimeMs = INVALID_TIME_MS;
private long _explicitlySetTimeBoundaryMs = INVALID_TIME_MS;
private volatile TimeBoundaryInfo _timeBoundaryInfo;
@@ -120,14 +124,12 @@ public class TimeBoundaryManager {
segmentZKMetadataPaths.add(_segmentZKMetadataPathPrefix + segment);
}
List<ZNRecord> znRecords = _propertyStore.get(segmentZKMetadataPaths,
null, AccessOption.PERSISTENT, false);
- long maxEndTimeMs = INVALID_TIME_MS;
for (int i = 0; i < numSegments; i++) {
String segment = segments.get(i);
- long endTimeMs = extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
znRecords.get(i));
- _endTimeMsMap.put(segment, endTimeMs);
- maxEndTimeMs = Math.max(maxEndTimeMs, endTimeMs);
+ _endTimeMsMap.put(segment,
extractEndTimeMsFromSegmentZKMetadataZNRecord(segment, znRecords.get(i)));
}
- updateTimeBoundaryInfo(maxEndTimeMs);
+ _maxEndTimeMs = computeMaxEndTimeMs();
+ updateTimeBoundaryInfo(_maxEndTimeMs);
}
private void updateExplicitlySetTimeBoundary(IdealState idealState) {
@@ -206,20 +208,58 @@ public class TimeBoundaryManager {
Set<String> onlineSegments) {
updateExplicitlySetTimeBoundary(idealState);
+ // Read the ZK metadata of the new segments in one batch
+ List<String> newSegments = new ArrayList<>();
+ List<String> newSegmentZKMetadataPaths = new ArrayList<>();
for (String segment : onlineSegments) {
// NOTE: Only update the segment end time when there are ONLINE
instances in the external view to prevent moving
// the time boundary before the new segment is picked up by the
servers
- Map<String, String> instanceStateMap = externalView.getStateMap(segment);
- if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
- _endTimeMsMap.computeIfAbsent(segment, k ->
extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
- _propertyStore.get(_segmentZKMetadataPathPrefix + segment, null,
AccessOption.PERSISTENT)));
+ if (!_endTimeMsMap.containsKey(segment)) {
+ Map<String, String> instanceStateMap =
externalView.getStateMap(segment);
+ if (instanceStateMap != null &&
instanceStateMap.containsValue(SegmentStateModel.ONLINE)) {
+ newSegments.add(segment);
+ newSegmentZKMetadataPaths.add(_segmentZKMetadataPathPrefix +
segment);
+ }
+ }
+ }
+ // NOTE: Read before changing any state, so that a failed read leaves the
state as it was. The read throws on a ZK
+ // error other than a missing node, so that the routing entry keeps
its IS/EV versions and retries.
+ List<ZNRecord> znRecords = newSegments.isEmpty() ? List.of()
+ : _propertyStore.get(newSegmentZKMetadataPaths, null,
AccessOption.PERSISTENT, true);
+ boolean recomputeMaxEndTime = false;
+ // NOTE: Removing a segment can only lower the max end time, so a full
scan is needed only when a removed segment
+ // holds the current max end time.
+ Iterator<Map.Entry<String, Long>> iterator =
_endTimeMsMap.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<String, Long> entry = iterator.next();
+ if (!onlineSegments.contains(entry.getKey())) {
+ if (entry.getValue() >= _maxEndTimeMs) {
+ recomputeMaxEndTime = true;
+ }
+ iterator.remove();
+ }
+ }
+ if (recomputeMaxEndTime) {
+ _maxEndTimeMs = computeMaxEndTimeMs();
+ }
+ int numNewSegments = newSegments.size();
+ for (int i = 0; i < numNewSegments; i++) {
+ String segment = newSegments.get(i);
+ ZNRecord znRecord = znRecords.get(i);
+ if (znRecord == null) {
+ // NOTE: The node does not exist. Do not record the segment, so that
the next assignment change reads it again.
+ LOGGER.warn("Failed to find segment ZK metadata for segment: {},
table: {}, will retry on next change", segment,
+ _offlineTableName);
+ continue;
}
+ long endTimeMs = extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
znRecord);
+ _endTimeMsMap.put(segment, endTimeMs);
+ _maxEndTimeMs = Math.max(_maxEndTimeMs, endTimeMs);
}
- _endTimeMsMap.keySet().retainAll(onlineSegments);
- updateTimeBoundaryInfo(getMaxEndTimeMs());
+ updateTimeBoundaryInfo(_maxEndTimeMs);
}
- private long getMaxEndTimeMs() {
+ private long computeMaxEndTimeMs() {
long maxEndTimeMs = INVALID_TIME_MS;
for (long endTimeMs : _endTimeMsMap.values()) {
maxEndTimeMs = Math.max(maxEndTimeMs, endTimeMs);
@@ -229,9 +269,16 @@ public class TimeBoundaryManager {
/// Refreshes the metadata for the given segment (called when segment is
getting refreshed).
public synchronized void refreshSegment(String segment) {
- _endTimeMsMap.put(segment,
extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
- _propertyStore.get(_segmentZKMetadataPathPrefix + segment, null,
AccessOption.PERSISTENT)));
- updateTimeBoundaryInfo(getMaxEndTimeMs());
+ long endTimeMs = extractEndTimeMsFromSegmentZKMetadataZNRecord(segment,
+ _propertyStore.get(_segmentZKMetadataPathPrefix + segment, null,
AccessOption.PERSISTENT));
+ Long previousEndTimeMs = _endTimeMsMap.put(segment, endTimeMs);
+ if (previousEndTimeMs != null && previousEndTimeMs >= _maxEndTimeMs &&
endTimeMs < previousEndTimeMs) {
+ // The refreshed segment held the max end time and moved it back
+ _maxEndTimeMs = computeMaxEndTimeMs();
+ } else {
+ _maxEndTimeMs = Math.max(_maxEndTimeMs, endTimeMs);
+ }
+ updateTimeBoundaryInfo(_maxEndTimeMs);
}
@Nullable
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
index 9901b71c96e..5c6d6af0cbb 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionMetadataManagerTest.java
@@ -19,6 +19,9 @@
package org.apache.pinot.broker.routing.segmentpartition;
import com.google.common.collect.ImmutableSet;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -36,13 +39,17 @@ import
org.apache.pinot.broker.routing.segmentmetadata.SegmentZkMetadataFetcher;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.TestClock;
import org.apache.pinot.controller.helix.ControllerTest;
+import org.apache.pinot.core.routing.TablePartitionInfo;
import org.apache.pinot.core.routing.TablePartitionReplicatedServersInfo;
import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
+import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.CONSUMING;
+import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.OFFLINE;
import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.ONLINE;
import static org.testng.Assert.*;
@@ -431,6 +438,132 @@ public class SegmentPartitionMetadataManagerTest extends
ControllerTest {
+
tablePartitionReplicatedServersInfo.getPartitionsWithOnlyDeferredSegments());
}
+ /// An assignment change that changes neither the segments, their partitions
nor their online servers must keep the
+ /// published objects, while a change of the online servers must only
replace the replicated servers info.
+ @Test
+ public void testRecomputeOnlyOnChange() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+ // NOTE: Ideal state is not used in the current implementation.
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ TimeUnit.MINUTES.toMillis(5));
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+
+ String segment0 = "recomputeSegment0";
+ String segment1 = "recomputeSegment1";
+ String segmentWithoutMetadata = "recomputeSegmentWithoutMetadata";
+ onlineSegments.add(segment0);
+ onlineSegments.add(segment1);
+ segmentAssignment.put(segment0, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ segmentAssignment.put(segment1, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ setSegmentZKMetadata(segment0, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ setSegmentZKMetadata(segment1, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 1,
0L);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+ // A segment without ZK metadata is pulled again on every change, which
alone must not count as a change
+ onlineSegments.add(segmentWithoutMetadata);
+ segmentAssignment.put(segmentWithoutMetadata, Map.of(SERVER_0, ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ TablePartitionInfo tablePartitionInfo =
partitionMetadataManager.getTablePartitionInfo();
+ TablePartitionReplicatedServersInfo replicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ assertEquals(replicatedServersInfo.getSegmentsWithInvalidPartition(),
List.of(segmentWithoutMetadata));
+
+ // No-op change, with a fresh but equal instance state map and an
unrelated state change in the external view
+ segmentAssignment.put(segment0, new HashMap<>(Map.of(SERVER_0, ONLINE,
SERVER_1, ONLINE)));
+ segmentAssignment.put("segmentNotOnline", Map.of(SERVER_0, ONLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
+ // An OFFLINE replica is not an online server
+ segmentAssignment.put(segment1, Map.of(SERVER_0, ONLINE, SERVER_1,
OFFLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+ replicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
replicatedServersInfo.getPartitionInfoMap();
+ assertEquals(partitionInfoMap[0]._fullyReplicatedServers, Set.of(SERVER_0,
SERVER_1));
+ assertEquals(partitionInfoMap[1]._fullyReplicatedServers,
Set.of(SERVER_0));
+
+ // Same online servers with a different state (CONSUMING instead of
ONLINE) is a no-op
+ segmentAssignment.put(segment1, Map.of(SERVER_0, CONSUMING, SERVER_1,
OFFLINE));
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
+ // Removing a segment replaces both objects
+ onlineSegments.remove(segment1);
+ segmentAssignment.remove(segment1);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ assertNotSame(partitionMetadataManager.getTablePartitionInfo(),
tablePartitionInfo);
+
assertNotSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
assertEquals(partitionMetadataManager.getTablePartitionInfo().getSegmentsByPartition(),
+ List.of(List.of(segment0), List.of()));
+
assertNull(partitionMetadataManager.getTablePartitionReplicatedServersInfo().getPartitionInfoMap()[1]);
+ }
+
+ /// The partition info re-evaluates which segments are new only when it is
recomputed, so an assignment change that
+ /// changes nothing else must still recompute once a new segment excluded
from the partition info is no longer new.
+ @Test
+ public void testNewSegmentExpiresOnNoOpChange() {
+ ExternalView externalView = new ExternalView(OFFLINE_TABLE_NAME);
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Set<String> onlineSegments = new HashSet<>();
+ // NOTE: Ideal state is not used in the current implementation.
+ IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
+
+ long newSegmentExpirationMs = TimeUnit.MINUTES.toMillis(5);
+ TestClock clock = new TestClock(Instant.now(), ZoneId.systemDefault());
+ SegmentPartitionMetadataManager partitionMetadataManager =
+ new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC, NUM_PARTITIONS,
+ newSegmentExpirationMs, clock);
+ SegmentZkMetadataFetcher segmentZkMetadataFetcher =
+ new SegmentZkMetadataFetcher(OFFLINE_TABLE_NAME, _propertyStore);
+ segmentZkMetadataFetcher.register(partitionMetadataManager);
+
+ String oldSegment = "expiryOldSegment";
+ onlineSegments.add(oldSegment);
+ segmentAssignment.put(oldSegment, Map.of(SERVER_0, ONLINE, SERVER_1,
ONLINE));
+ setSegmentZKMetadata(oldSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
+ segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
+
+ // A new segment missing one replica is excluded from the partition info.
It stops being new 1 second from now.
+ String newSegment = "expiryNewSegment";
+ long creationTimeMs = clock.millis() - newSegmentExpirationMs + 1000;
+ onlineSegments.add(newSegment);
+ segmentAssignment.put(newSegment, Map.of(SERVER_0, ONLINE));
+ setSegmentZKMetadata(newSegment, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
creationTimeMs);
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ TablePartitionReplicatedServersInfo replicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo.PartitionInfo partitionInfo =
replicatedServersInfo.getPartitionInfoMap()[0];
+ assertEquals(partitionInfo._segments, List.of(oldSegment));
+ assertEquals(partitionInfo._fullyReplicatedServers, Set.of(SERVER_0,
SERVER_1));
+
+ // While the segment is still new, a no-op change keeps the published
object
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+
+ // Move the clock past the expiration of the segment
+ clock.fastForward(Duration.ofSeconds(2));
+
+ // A no-op change must now include the segment, as a regular segment
reducing the fully replicated servers
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+ replicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
+ partitionInfo = replicatedServersInfo.getPartitionInfoMap()[0];
+ assertEqualsNoOrder(partitionInfo._segments.toArray(), new
String[]{oldSegment, newSegment});
+ assertEquals(partitionInfo._fullyReplicatedServers, Set.of(SERVER_0));
+
+ // Once no segment is new, a no-op change keeps the published object
+ segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
+
assertSame(partitionMetadataManager.getTablePartitionReplicatedServersInfo(),
replicatedServersInfo);
+ }
+
private void setSegmentZKMetadata(String segment, String partitionFunction,
int numPartitions, int partitionId,
long creationTimeMs) {
SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segment);
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtilsTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtilsTest.java
new file mode 100644
index 00000000000..9dd3a1130be
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/segmentpartition/SegmentPartitionUtilsTest.java
@@ -0,0 +1,108 @@
+/**
+ * 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.pinot.broker.routing.segmentpartition;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.segment.spi.partition.metadata.ColumnPartitionMetadata;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNotSame;
+import static org.testng.Assert.assertSame;
+
+
+/// Tests for [SegmentPartitionUtils].
+public class SegmentPartitionUtilsTest {
+ private static final String TABLE_NAME = "testTable_OFFLINE";
+ private static final String PARTITION_COLUMN = "memberId";
+ private static final String OTHER_PARTITION_COLUMN = "groupId";
+
+ /// Segments with equal partition function metadata share one partition
function, and one-element partition sets
+ /// are shared too.
+ @Test
+ public void testSharedPartitionFunctionsAndPartitionSets() {
+ SegmentPartitionInfo info0 = extract("segment0", "Murmur", 8, Set.of(3),
null);
+ SegmentPartitionInfo info1 = extract("segment1", "Murmur", 8, Set.of(3),
null);
+ SegmentPartitionInfo info2 = extract("segment2", "Murmur", 8, Set.of(5),
null);
+ assertSame(info1.getPartitionFunction(), info0.getPartitionFunction());
+ assertSame(info2.getPartitionFunction(), info0.getPartitionFunction());
+ assertSame(info1.getPartitions(), info0.getPartitions());
+ assertEquals(info0.getPartitions(), Set.of(3));
+ assertEquals(info2.getPartitions(), Set.of(5));
+ assertEquals(info0.getPartitionFunction().getName(), "Murmur");
+ assertEquals(info0.getPartitionFunction().getNumPartitions(), 8);
+
+ // A different number of partitions, function or function config is a
different function
+ SegmentPartitionInfo otherNumPartitions = extract("segment3", "Murmur",
16, Set.of(3), null);
+ assertNotSame(otherNumPartitions.getPartitionFunction(),
info0.getPartitionFunction());
+ assertEquals(otherNumPartitions.getPartitionFunction().getNumPartitions(),
16);
+ SegmentPartitionInfo otherFunction = extract("segment4", "Modulo", 8,
Set.of(3), null);
+ assertNotSame(otherFunction.getPartitionFunction(),
info0.getPartitionFunction());
+ assertEquals(otherFunction.getPartitionFunction().getName(), "Modulo");
+ SegmentPartitionInfo otherConfig = extract("segment5", "Murmur", 8,
Set.of(3), Map.of("useRawBytes", "true"));
+ assertNotSame(otherConfig.getPartitionFunction(),
info0.getPartitionFunction());
+
+ // Multi-partition sets are kept as is
+ SegmentPartitionInfo multiPartitions = extract("segment6", "Murmur", 8,
Set.of(1, 2), null);
+ assertEquals(multiPartitions.getPartitions(), Set.of(1, 2));
+ assertSame(multiPartitions.getPartitionFunction(),
info0.getPartitionFunction());
+ }
+
+ @Test
+ public void testSharedPartitionFunctionsInPartitionInfoMap() {
+ ZNRecord znRecord0 = znRecord("segment0", "Murmur", 8, Set.of(3), null);
+ ZNRecord znRecord1 = znRecord("segment1", "Murmur", 8, Set.of(3), null);
+ Set<String> partitionColumns = Set.of(PARTITION_COLUMN,
OTHER_PARTITION_COLUMN);
+ Map<String, SegmentPartitionInfo> infoMap0 =
+ SegmentPartitionUtils.extractPartitionInfoMap(TABLE_NAME,
partitionColumns, "segment0", znRecord0);
+ Map<String, SegmentPartitionInfo> infoMap1 =
+ SegmentPartitionUtils.extractPartitionInfoMap(TABLE_NAME,
partitionColumns, "segment1", znRecord1);
+ assertNotNull(infoMap0);
+ assertNotNull(infoMap1);
+ for (String partitionColumn : partitionColumns) {
+ assertSame(infoMap1.get(partitionColumn).getPartitionFunction(),
+ infoMap0.get(partitionColumn).getPartitionFunction());
+ assertSame(infoMap1.get(partitionColumn).getPartitions(),
infoMap0.get(partitionColumn).getPartitions());
+ assertEquals(infoMap0.get(partitionColumn).getPartitions(), Set.of(3));
+ }
+ }
+
+ private static SegmentPartitionInfo extract(String segment, String
functionName, int numPartitions,
+ Set<Integer> partitions, Map<String, String> functionConfig) {
+ SegmentPartitionInfo info =
SegmentPartitionUtils.extractPartitionInfo(TABLE_NAME, PARTITION_COLUMN,
segment,
+ znRecord(segment, functionName, numPartitions, partitions,
functionConfig));
+ assertNotNull(info);
+ return info;
+ }
+
+ private static ZNRecord znRecord(String segment, String functionName, int
numPartitions, Set<Integer> partitions,
+ Map<String, String> functionConfig) {
+ ColumnPartitionMetadata columnPartitionMetadata =
+ new ColumnPartitionMetadata(functionName, numPartitions, partitions,
functionConfig);
+ SegmentZKMetadata segmentZKMetadata = new SegmentZKMetadata(segment);
+ segmentZKMetadata.setPartitionMetadata(new SegmentPartitionMetadata(
+ Map.of(PARTITION_COLUMN, columnPartitionMetadata,
OTHER_PARTITION_COLUMN, columnPartitionMetadata)));
+ return segmentZKMetadata.toZNRecord();
+ }
+}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManagerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManagerTest.java
index 694915ebcff..b61cbd89ae6 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManagerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/timeboundary/TimeBoundaryManagerTest.java
@@ -19,9 +19,11 @@
package org.apache.pinot.broker.routing.timeboundary;
import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import org.apache.helix.api.exceptions.HelixMetaDataAccessException;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
@@ -52,6 +54,7 @@ import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.Segmen
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
public class TimeBoundaryManagerTest extends ControllerTest {
@@ -226,6 +229,124 @@ public class TimeBoundaryManagerTest extends
ControllerTest {
verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(),
expectedTimeValue);
}
+ /// The ZK metadata of the segments added by an assignment change must be
read in one batch, and removing a segment
+ /// must only move the time boundary back when that segment held the max end
time.
+ @Test
+ public void testBatchedReadsAndIncrementalMaxEndTime() {
+ String rawTableName = "testTableBatchedReads";
+ TableConfig tableConfig = getTableConfig(rawTableName, "DAILY");
+ TimeUnit timeUnit = TimeUnit.DAYS;
+ setSchemaTimeFieldSpec(rawTableName, timeUnit);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = Mockito.spy(_propertyStore);
+ TimeBoundaryManager timeBoundaryManager =
+ new TimeBoundaryManager(tableConfig, propertyStore,
Mockito.mock(BrokerMetrics.class));
+
+ ExternalView externalView = new ExternalView(tableConfig.getTableName());
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Map<String, String> onlineInstanceStateMap = Map.of("server", ONLINE);
+ Set<String> onlineSegments = new HashSet<>();
+ IdealState idealState = new IdealState("");
+ String segment0 = "batchedSegment0";
+ onlineSegments.add(segment0);
+ segmentAssignment.put(segment0, onlineInstanceStateMap);
+ setSegmentZKMetadata(rawTableName, segment0, 3, timeUnit);
+ timeBoundaryManager.init(idealState, externalView, onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 2);
+
+ // Adding 2 segments reads their ZK metadata in one batch
+ String segment1 = "batchedSegment1";
+ String segment2 = "batchedSegment2";
+ setSegmentZKMetadata(rawTableName, segment1, 4, timeUnit);
+ setSegmentZKMetadata(rawTableName, segment2, 6, timeUnit);
+ onlineSegments.add(segment1);
+ onlineSegments.add(segment2);
+ segmentAssignment.put(segment1, onlineInstanceStateMap);
+ segmentAssignment.put(segment2, onlineInstanceStateMap);
+ Mockito.clearInvocations(propertyStore);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 5);
+ Mockito.verify(propertyStore).get(Mockito.argThat((List<String> paths) ->
paths.size() == 2), Mockito.any(),
+ Mockito.anyInt(), Mockito.anyBoolean());
+ Mockito.verifyNoMoreInteractions(propertyStore);
+
+ // An assignment change without new segment does not read ZK metadata
+ Mockito.clearInvocations(propertyStore);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 5);
+ Mockito.verifyNoInteractions(propertyStore);
+
+ // Removing a segment that does not hold the max end time keeps the time
boundary
+ onlineSegments.remove(segment1);
+ segmentAssignment.remove(segment1);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 5);
+
+ // Removing the segment that holds the max end time moves the time
boundary back
+ onlineSegments.remove(segment2);
+ segmentAssignment.remove(segment2);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 2);
+ Mockito.verifyNoInteractions(propertyStore);
+
+ // Refreshing the segment holding the max end time to a smaller end time
moves the time boundary back
+ setSegmentZKMetadata(rawTableName, segment0, 2, timeUnit);
+ timeBoundaryManager.refreshSegment(segment0);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 1);
+
+ // A new segment whose ZK metadata cannot be read is read again on the
next change
+ String segment3 = "batchedSegment3";
+ onlineSegments.add(segment3);
+ segmentAssignment.put(segment3, onlineInstanceStateMap);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 1);
+ setSegmentZKMetadata(rawTableName, segment3, 8, timeUnit);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 7);
+ }
+
+ /// A ZK read failure other than a missing node must propagate, and must
leave the time boundary and the state as they
+ /// were, so that the next assignment change can replace the segments.
+ @Test
+ public void testReadFailureDuringReplacementKeepsTimeBoundary() {
+ String rawTableName = "testTableReadFailure";
+ TableConfig tableConfig = getTableConfig(rawTableName, "DAILY");
+ TimeUnit timeUnit = TimeUnit.DAYS;
+ setSchemaTimeFieldSpec(rawTableName, timeUnit);
+ ZkHelixPropertyStore<ZNRecord> propertyStore = Mockito.spy(_propertyStore);
+ TimeBoundaryManager timeBoundaryManager =
+ new TimeBoundaryManager(tableConfig, propertyStore,
Mockito.mock(BrokerMetrics.class));
+
+ ExternalView externalView = new ExternalView(tableConfig.getTableName());
+ Map<String, Map<String, String>> segmentAssignment =
externalView.getRecord().getMapFields();
+ Map<String, String> onlineInstanceStateMap = Map.of("server", ONLINE);
+ Set<String> onlineSegments = new HashSet<>();
+ IdealState idealState = new IdealState("");
+ String oldSegment = "replacedSegment";
+ onlineSegments.add(oldSegment);
+ segmentAssignment.put(oldSegment, onlineInstanceStateMap);
+ setSegmentZKMetadata(rawTableName, oldSegment, 3, timeUnit);
+ timeBoundaryManager.init(idealState, externalView, onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 2);
+
+ // Replace the old segment with a new one, and fail the read of the new
segment ZK metadata
+ String newSegment = "replacingSegment";
+ setSegmentZKMetadata(rawTableName, newSegment, 5, timeUnit);
+ onlineSegments.remove(oldSegment);
+ segmentAssignment.remove(oldSegment);
+ onlineSegments.add(newSegment);
+ segmentAssignment.put(newSegment, onlineInstanceStateMap);
+ Mockito.doThrow(new HelixMetaDataAccessException("Failed to read node,
return code: OPERATIONTIMEOUT"))
+ .when(propertyStore).get(Mockito.anyList(), Mockito.any(),
Mockito.anyInt(), Mockito.eq(true));
+ assertThrows(HelixMetaDataAccessException.class,
+ () -> timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments));
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 2);
+
+ // The next assignment change reads the new segment again and replaces the
old one
+ Mockito.reset(propertyStore);
+ timeBoundaryManager.onAssignmentChange(idealState, externalView,
onlineSegments);
+ verifyTimeBoundaryInfo(timeBoundaryManager.getTimeBoundaryInfo(), 4);
+ }
+
private TableConfig getTableConfig(String rawTableName, String
pushFrequency) {
return new
TableConfigBuilder(TableType.OFFLINE).setTableName(rawTableName).setTimeColumnName(TIME_COLUMN)
.setSegmentPushFrequency(pushFrequency).build();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
