This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 0f5a71e2c65 Make new segment expiration time configurable in
SegmentPartitionMetadataManager (#18707)
0f5a71e2c65 is described below
commit 0f5a71e2c65cba8a16e7359f5c53c2e7de9cc5ca
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Mon Jun 8 17:23:16 2026 -0700
Make new segment expiration time configurable in
SegmentPartitionMetadataManager (#18707)
---
.../broker/routing/instanceselector/InstanceSelector.java | 7 -------
.../broker/routing/manager/BaseBrokerRoutingManager.java | 10 +++++++---
.../segmentpartition/SegmentPartitionMetadataManager.java | 6 ++++--
.../routing/instanceselector/InstanceSelectorTest.java | 10 +++++++---
.../SegmentPartitionMetadataManagerTest.java | 13 +++++++------
5 files changed, 25 insertions(+), 21 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java
index 25a54cd7fd7..664ea846368 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelector.java
@@ -22,7 +22,6 @@ import java.time.Clock;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.helix.model.ExternalView;
@@ -41,12 +40,6 @@ import org.apache.pinot.spi.config.table.TableConfig;
* The instance selector selects server instances to serve the query based on
the selected segments.
*/
public interface InstanceSelector {
- long NEW_SEGMENT_EXPIRATION_MILLIS = TimeUnit.MINUTES.toMillis(5);
-
- static boolean isNewSegment(long creationTimeMs, long currentTimeMs) {
- return isNewSegment(creationTimeMs, currentTimeMs,
NEW_SEGMENT_EXPIRATION_MILLIS);
- }
-
static boolean isNewSegment(long creationTimeMs, long currentTimeMs, long
newSegmentExpirationMillis) {
return creationTimeMs > 0 && currentTimeMs - creationTimeMs <=
newSegmentExpirationMillis;
}
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
index f984137546f..13adc526656 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java
@@ -138,6 +138,7 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
private final ServerRoutingStatsManager _serverRoutingStatsManager;
private final PinotConfiguration _pinotConfig;
private final boolean _enablePartitionMetadataManager;
+ private final long _newSegmentExpirationMs;
private final ExecutorService _executorService;
@Nullable
private Consumer<ServerInstance> _serverReenableCallback;
@@ -180,6 +181,9 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
_enablePartitionMetadataManager =
pinotConfig.getProperty(CommonConstants.Broker.CONFIG_OF_ENABLE_PARTITION_METADATA_MANAGER,
CommonConstants.Broker.DEFAULT_ENABLE_PARTITION_METADATA_MANAGER);
+ _newSegmentExpirationMs = TimeUnit.SECONDS.toMillis(
+
pinotConfig.getProperty(CommonConstants.Broker.CONFIG_OF_NEW_SEGMENT_EXPIRATION_SECONDS,
+
CommonConstants.Broker.DEFAULT_VALUE_OF_NEW_SEGMENT_EXPIRATION_SECONDS));
int processSegmentAssignmentChangeNumThreads =
pinotConfig.getProperty(CommonConstants.Broker.CONFIG_OF_ROUTING_ASSIGNMENT_CHANGE_PROCESS_PARALLELISM,
CommonConstants.Broker.DEFAULT_ROUTING_ASSIGNMENT_CHANGE_PROCESS_PARALLELISM);
@@ -807,9 +811,9 @@ public abstract class BaseBrokerRoutingManager implements
RoutingManager, Cluste
columnPartitionMap.entrySet().iterator().next();
LOGGER.info("Enabling SegmentPartitionMetadataManager for table:
{} on partition column: {}",
tableNameWithType, partitionConfig.getKey());
- partitionMetadataManager =
- new SegmentPartitionMetadataManager(tableNameWithType,
partitionConfig.getKey(),
- partitionConfig.getValue().getFunctionName(),
partitionConfig.getValue().getNumPartitions());
+ partitionMetadataManager = new
SegmentPartitionMetadataManager(tableNameWithType, partitionConfig.getKey(),
+ partitionConfig.getValue().getFunctionName(),
partitionConfig.getValue().getNumPartitions(),
+ _newSegmentExpirationMs);
} else {
LOGGER.warn(
"Cannot enable SegmentPartitionMetadataManager for table: {}
with multiple partition columns: {}",
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 6abc5864ae1..68c95071c45 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
@@ -63,6 +63,7 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
private final String _partitionColumn;
private final String _partitionFunctionName;
private final int _numPartitions;
+ private final long _newSegmentExpirationMs;
// cache-able content, only follow changes if onlineSegments list (of
ideal-state) is changed.
private final Map<String, SegmentInfo> _segmentInfoMap = new HashMap<>();
@@ -72,11 +73,12 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
private transient TablePartitionReplicatedServersInfo
_tablePartitionReplicatedServersInfo;
public SegmentPartitionMetadataManager(String tableNameWithType, String
partitionColumn, String partitionFunctionName,
- int numPartitions) {
+ int numPartitions, long newSegmentExpirationMs) {
_tableNameWithType = tableNameWithType;
_partitionColumn = partitionColumn;
_partitionFunctionName = partitionFunctionName;
_numPartitions = numPartitions;
+ _newSegmentExpirationMs = newSegmentExpirationMs;
}
@Override
@@ -215,7 +217,7 @@ public class SegmentPartitionMetadataManager implements
SegmentZkMetadataFetchLi
continue;
}
// Process new segments in the end
- if (InstanceSelector.isNewSegment(segmentInfo._creationTimeMs,
currentTimeMs)) {
+ if (InstanceSelector.isNewSegment(segmentInfo._creationTimeMs,
currentTimeMs, _newSegmentExpirationMs)) {
newSegmentInfoEntries.add(entry);
continue;
}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java
index 668332e0f68..ca7a66dad5d 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/routing/instanceselector/InstanceSelectorTest.java
@@ -59,7 +59,6 @@ import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
-import static
org.apache.pinot.broker.routing.instanceselector.InstanceSelector.NEW_SEGMENT_EXPIRATION_MILLIS;
import static
org.apache.pinot.spi.config.table.RoutingConfig.REPLICA_GROUP_INSTANCE_SELECTOR_TYPE;
import static
org.apache.pinot.spi.config.table.RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE;
import static
org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel.CONSUMING;
@@ -70,7 +69,9 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.clearInvocations;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
@@ -109,7 +110,10 @@ public class InstanceSelectorTest {
private final static List<String> SEGMENTS =
Arrays.asList("segment0", "segment1", "segment2", "segment3",
"segment4", "segment5", "segment6", "segment7",
"segment8", "segment9", "segment10", "segment11");
- private final static InstanceSelectorConfig INSTANCE_SELECTOR_CONFIG = new
InstanceSelectorConfig(false, 300, false);
+ private static final long NEW_SEGMENT_EXPIRATION_SECONDS = 300;
+ private static final long NEW_SEGMENT_EXPIRATION_MILLIS =
TimeUnit.SECONDS.toMillis(NEW_SEGMENT_EXPIRATION_SECONDS);
+ private final static InstanceSelectorConfig INSTANCE_SELECTOR_CONFIG =
+ new InstanceSelectorConfig(false, NEW_SEGMENT_EXPIRATION_SECONDS, false);
private void createSegments(List<Pair<String, Long>>
segmentCreationTimeMsPairs) {
List<String> segmentZKMetadataPaths = new ArrayList<>();
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 f80f32c62ef..5583172b164 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
@@ -23,6 +23,7 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.IdealState;
@@ -83,16 +84,16 @@ public class SegmentPartitionMetadataManagerTest extends
ControllerTest {
IdealState idealState = new IdealState(OFFLINE_TABLE_NAME);
SegmentPartitionMetadataManager partitionMetadataManager =
- new SegmentPartitionMetadataManager(OFFLINE_TABLE_NAME,
PARTITION_COLUMN, PARTITION_COLUMN_FUNC,
- NUM_PARTITIONS);
+ 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);
// Initial state should be all empty
segmentZkMetadataFetcher.init(idealState, externalView, onlineSegments);
- TablePartitionReplicatedServersInfo tablePartitionReplicatedServersInfo =
partitionMetadataManager
- .getTablePartitionReplicatedServersInfo();
+ TablePartitionReplicatedServersInfo tablePartitionReplicatedServersInfo =
+ partitionMetadataManager.getTablePartitionReplicatedServersInfo();
assertEquals(tablePartitionReplicatedServersInfo.getPartitionInfoMap(),
new TablePartitionReplicatedServersInfo.PartitionInfo[NUM_PARTITIONS]);
assertTrue(tablePartitionReplicatedServersInfo.getSegmentsWithInvalidPartition().isEmpty());
@@ -149,8 +150,8 @@ public class SegmentPartitionMetadataManagerTest extends
ControllerTest {
setSegmentZKMetadata(segment0, PARTITION_COLUMN_FUNC, NUM_PARTITIONS, 0,
0L);
segmentZkMetadataFetcher.onAssignmentChange(idealState, externalView,
onlineSegments);
tablePartitionReplicatedServersInfo =
partitionMetadataManager.getTablePartitionReplicatedServersInfo();
- TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
tablePartitionReplicatedServersInfo
- .getPartitionInfoMap();
+ TablePartitionReplicatedServersInfo.PartitionInfo[] partitionInfoMap =
+ tablePartitionReplicatedServersInfo.getPartitionInfoMap();
assertEquals(partitionInfoMap[0]._fullyReplicatedServers,
Collections.singleton(SERVER_0));
assertEquals(partitionInfoMap[0]._segments,
Collections.singleton(segment0));
assertNull(partitionInfoMap[1]);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
