You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/04/15 11:31:42 UTC
(pinot) 24/25: Always emit COMMITTING_SEGMENT_SIZE and NUM_ROWS_THRESHOLD regardless of the threshold update type (#18137) (#501)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch release/1.5.0-STP-2.134.x
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 4aa0f7695b0c588403bdd0133fbc19e49e90bd21
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Thu Apr 9 14:45:50 2026 -0700
Always emit COMMITTING_SEGMENT_SIZE and NUM_ROWS_THRESHOLD regardless of the threshold update type (#18137) (#501)
---
.../realtime/PinotLLCRealtimeSegmentManager.java | 50 ++-
.../segment/CommittingSegmentDescriptor.java | 23 +-
.../segment/FlushThresholdUpdateManager.java | 4 +-
.../SegmentSizeBasedFlushThresholdUpdater.java | 19 +-
.../PinotLLCRealtimeSegmentManagerTest.java | 409 +++++++++------------
.../segment/FlushThresholdUpdaterTest.java | 15 +-
6 files changed, 232 insertions(+), 288 deletions(-)
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
index 7913e88c07c..b6630828feb 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java
@@ -607,8 +607,7 @@ public class PinotLLCRealtimeSegmentManager {
// Copy the segment file to the controller
String segmentLocation = committingSegmentDescriptor.getSegmentLocation();
Preconditions.checkArgument(segmentLocation != null, "Segment location must be provided");
- if (segmentLocation.regionMatches(true, 0, CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME, 0,
- CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length())) {
+ if (isPeerUrl(segmentLocation)) {
LOGGER.info("No moving needed for segment on peer servers: {}", segmentLocation);
return;
}
@@ -632,6 +631,11 @@ public class PinotLLCRealtimeSegmentManager {
committingSegmentDescriptor.setSegmentLocation(uriToMoveTo);
}
+ private boolean isPeerUrl(String segmentLocation) {
+ return segmentLocation.regionMatches(true, 0, CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME, 0,
+ CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
+ }
+
/**
* This method is invoked after the realtime segment is uploaded but before a response is sent to the server.
* It updates the propertystore segment metadata from IN_PROGRESS to DONE, and also creates new propertystore
@@ -671,9 +675,9 @@ public class PinotLLCRealtimeSegmentManager {
// Step-1: Update PROPERTYSTORE
LOGGER.info("Committing segment metadata for segment: {}", committingSegmentName);
long startTimeNs1 = System.nanoTime();
- SegmentZKMetadata committingSegmentZKMetadata =
- toCommitting ? updateSegmentZKMetadataToCommitting(realtimeTableName, committingSegmentDescriptor)
- : updateSegmentZKMetadataToDone(realtimeTableName, committingSegmentDescriptor, Status.IN_PROGRESS);
+ SegmentZKMetadata committingSegmentZKMetadata = toCommitting
+ ? updateSegmentZKMetadataToCommitting(realtimeTableName, committingSegmentDescriptor)
+ : updateSegmentZKMetadataToDone(realtimeTableName, committingSegmentDescriptor, Status.IN_PROGRESS);
preProcessNewSegmentZKMetadata();
@@ -689,7 +693,7 @@ public class PinotLLCRealtimeSegmentManager {
LOGGER.info("Updating Idealstate for previous: {} and new segment: {}", committingSegmentName,
newConsumingSegmentName);
long startTimeNs3 = System.nanoTime();
- Map<String, Map<String, String>> instanceStatesMapAfterStep3 = Collections.emptyMap();
+ Map<String, Map<String, String>> instanceStatesMapAfterStep3;
boolean newConsumingSegmentInIdealState = false;
// When multiple segments of the same table complete around the same time it is possible that
@@ -788,8 +792,10 @@ public class PinotLLCRealtimeSegmentManager {
Preconditions.checkState(segmentMetadata != null, "Failed to find segment metadata from descriptor for segment: %s",
segmentName);
String segmentLocation = committingSegmentDescriptor.getSegmentLocation();
- String downloadUrl =
- isPeerURL(segmentLocation) ? CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD : segmentLocation;
+ Preconditions.checkArgument(segmentLocation != null, "Segment location must be provided");
+ String downloadUrl = isPeerUrl(segmentLocation)
+ ? CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD
+ : segmentLocation;
SegmentZKMetadataUtils.updateCommittingSegmentZKMetadata(realtimeTableName, segmentZKMetadata, segmentMetadata,
downloadUrl, committingSegmentDescriptor.getSegmentSizeBytes(), committingSegmentDescriptor.getNextOffset());
persistSegmentZKMetadata(realtimeTableName, segmentZKMetadata, stat.getVersion());
@@ -976,6 +982,7 @@ public class PinotLLCRealtimeSegmentManager {
FlushThresholdUpdater flushThresholdUpdater =
_flushThresholdUpdateManager.getFlushThresholdUpdater(streamConfig);
flushThresholdUpdater.onSegmentCommit(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata);
+ updateCommittingSegmentSizeGauge(streamConfig, committingSegmentDescriptor.getSegmentSizeBytes());
} catch (Exception e) {
LOGGER.error("Caught exception while updating flush threshold for table: {}, segment: {}", realtimeTableName,
segmentName, e);
@@ -985,11 +992,6 @@ public class PinotLLCRealtimeSegmentManager {
}
}
- private boolean isPeerURL(String segmentLocation) {
- return segmentLocation != null && segmentLocation.toLowerCase()
- .startsWith(CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME);
- }
-
/**
* Creates and persists segment ZK metadata for the new CONSUMING segment.
*/
@@ -1038,13 +1040,35 @@ public class PinotLLCRealtimeSegmentManager {
FlushThresholdUpdater flushThresholdUpdater = _flushThresholdUpdateManager.getFlushThresholdUpdater(streamConfig);
if (committingSegmentZKMetadata != null) {
flushThresholdUpdater.onSegmentCommit(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata);
+ long segmentSize = committingSegmentDescriptor.getSegmentSizeBytes();
+ // Segment size might not be available for pauseless ingestion
+ if (segmentSize > 0) {
+ updateCommittingSegmentSizeGauge(streamConfig, segmentSize);
+ }
}
flushThresholdUpdater.updateFlushThreshold(streamConfig, newSegmentZKMetadata,
getMaxNumPartitionsPerInstance(instancePartitions, numPartitions, numReplicas));
+ updateFlushThresholdGauge(streamConfig, newSegmentZKMetadata.getSizeThresholdToFlushSegment());
persistSegmentZKMetadata(realtimeTableName, newSegmentZKMetadata, -1);
}
+ private void updateCommittingSegmentSizeGauge(StreamConfig streamConfig, long segmentSize) {
+ String realtimeTableName = streamConfig.getTableNameWithType();
+ String topicName = streamConfig.getTopicName();
+ _controllerMetrics.setOrUpdateTableGauge(realtimeTableName, ControllerGauge.COMMITTING_SEGMENT_SIZE, segmentSize);
+ _controllerMetrics.setOrUpdateTableGauge(realtimeTableName, topicName,
+ ControllerGauge.COMMITTING_SEGMENT_SIZE_WITH_TOPIC, segmentSize);
+ }
+
+ private void updateFlushThresholdGauge(StreamConfig streamConfig, int flushThreshold) {
+ String realtimeTableName = streamConfig.getTableNameWithType();
+ String topicName = streamConfig.getTopicName();
+ _controllerMetrics.setOrUpdateTableGauge(realtimeTableName, ControllerGauge.NUM_ROWS_THRESHOLD, flushThreshold);
+ _controllerMetrics.setOrUpdateTableGauge(realtimeTableName, topicName,
+ ControllerGauge.NUM_ROWS_THRESHOLD_WITH_TOPIC, flushThreshold);
+ }
+
private String computeStartOffset(String nextOffset, StreamConfig streamConfig, int partitionId) {
if (!streamConfig.isEnableOffsetAutoReset() || streamConfig.isBackfillTopic()) {
return nextOffset;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/CommittingSegmentDescriptor.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/CommittingSegmentDescriptor.java
index ba89389c654..12c865bbcc7 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/CommittingSegmentDescriptor.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/CommittingSegmentDescriptor.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.helix.core.realtime.segment;
+import javax.annotation.Nullable;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
@@ -26,10 +27,11 @@ import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
* Class to hold properties of the committing segment
*/
public class CommittingSegmentDescriptor {
- private String _segmentName;
- private long _segmentSizeBytes;
+ private final String _segmentName;
+ private final String _nextOffset;
+ private final long _segmentSizeBytes;
+
private String _segmentLocation;
- private String _nextOffset;
private SegmentMetadataImpl _segmentMetadata;
private String _stopReason;
private int _preCommitRowCount;
@@ -70,18 +72,15 @@ public class CommittingSegmentDescriptor {
return _segmentName;
}
- public void setSegmentName(String segmentName) {
- _segmentName = segmentName;
+ public String getNextOffset() {
+ return _nextOffset;
}
public long getSegmentSizeBytes() {
return _segmentSizeBytes;
}
- public void setSegmentSizeBytes(long segmentSizeBytes) {
- _segmentSizeBytes = segmentSizeBytes;
- }
-
+ @Nullable
public String getSegmentLocation() {
return _segmentLocation;
}
@@ -90,10 +89,7 @@ public class CommittingSegmentDescriptor {
_segmentLocation = segmentLocation;
}
- public String getNextOffset() {
- return _nextOffset;
- }
-
+ @Nullable
public SegmentMetadataImpl getSegmentMetadata() {
return _segmentMetadata;
}
@@ -102,6 +98,7 @@ public class CommittingSegmentDescriptor {
_segmentMetadata = segmentMetadata;
}
+ @Nullable
public String getStopReason() {
return _stopReason;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java
index b4476e3bac1..119698f981b 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdateManager.java
@@ -47,7 +47,6 @@ public class FlushThresholdUpdateManager {
*/
public FlushThresholdUpdater getFlushThresholdUpdater(StreamConfig streamConfig) {
String tableTopicKey = getKey(streamConfig);
- String realtimeTableName = streamConfig.getTableNameWithType();
int flushThresholdRows = streamConfig.getFlushThresholdRows();
if (flushThresholdRows > 0) {
_flushThresholdUpdaterMap.remove(tableTopicKey);
@@ -61,8 +60,7 @@ public class FlushThresholdUpdateManager {
// Legacy behavior: when flush threshold rows is explicitly set to 0, use segment size based flush threshold
long flushThresholdSegmentSizeBytes = streamConfig.getFlushThresholdSegmentSizeBytes();
if (flushThresholdRows == 0 || flushThresholdSegmentSizeBytes > 0) {
- return _flushThresholdUpdaterMap.computeIfAbsent(tableTopicKey,
- k -> new SegmentSizeBasedFlushThresholdUpdater(realtimeTableName, streamConfig.getTopicName()));
+ return _flushThresholdUpdaterMap.computeIfAbsent(tableTopicKey, k -> new SegmentSizeBasedFlushThresholdUpdater());
} else {
_flushThresholdUpdaterMap.remove(tableTopicKey);
return new DefaultFlushThresholdUpdater(StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
index 4f76a3382f3..c3bdb0ac81d 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentSizeBasedFlushThresholdUpdater.java
@@ -19,8 +19,6 @@
package org.apache.pinot.controller.helix.core.realtime.segment;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.metrics.ControllerGauge;
-import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.spi.stream.StreamConfig;
@@ -34,25 +32,14 @@ import org.apache.pinot.spi.stream.StreamConfig;
*/
public class SegmentSizeBasedFlushThresholdUpdater implements FlushThresholdUpdater {
private final SizeBasedSegmentFlushThresholdComputer _flushThresholdComputer;
- private final String _realtimeTableName;
- private final String _topicName;
- private final ControllerMetrics _controllerMetrics = ControllerMetrics.get();
-
- public SegmentSizeBasedFlushThresholdUpdater(String realtimeTableName, String topicName) {
+ public SegmentSizeBasedFlushThresholdUpdater() {
_flushThresholdComputer = new SizeBasedSegmentFlushThresholdComputer();
- _realtimeTableName = realtimeTableName;
- _topicName = topicName;
}
@Override
public void onSegmentCommit(StreamConfig streamConfig, CommittingSegmentDescriptor committingSegmentDescriptor,
SegmentZKMetadata committingSegmentZKMetadata) {
- long segmentSize = committingSegmentDescriptor.getSegmentSizeBytes();
- _controllerMetrics.setOrUpdateTableGauge(_realtimeTableName, ControllerGauge.COMMITTING_SEGMENT_SIZE, segmentSize);
- _controllerMetrics.setOrUpdateTableGauge(_realtimeTableName, _topicName,
- ControllerGauge.COMMITTING_SEGMENT_SIZE_WITH_TOPIC, segmentSize);
-
_flushThresholdComputer.onSegmentCommit(committingSegmentDescriptor, committingSegmentZKMetadata);
}
@@ -61,9 +48,5 @@ public class SegmentSizeBasedFlushThresholdUpdater implements FlushThresholdUpda
int maxNumPartitionsPerInstance) {
int threshold = _flushThresholdComputer.computeThreshold(streamConfig, newSegmentZKMetadata.getSegmentName());
newSegmentZKMetadata.setSizeThresholdToFlushSegment(threshold);
-
- _controllerMetrics.setOrUpdateTableGauge(_realtimeTableName, ControllerGauge.NUM_ROWS_THRESHOLD, threshold);
- _controllerMetrics.setOrUpdateTableGauge(_realtimeTableName, _topicName,
- ControllerGauge.NUM_ROWS_THRESHOLD_WITH_TOPIC, threshold);
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
index eecad3d22c4..eab937c82f0 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java
@@ -124,21 +124,22 @@ public class PinotLLCRealtimeSegmentManagerTest {
private static final String CLUSTER_NAME = "testCluster";
private static final String RAW_TABLE_NAME = "testTable";
private static final String REALTIME_TABLE_NAME = TableNameBuilder.REALTIME.tableNameWithType(RAW_TABLE_NAME);
-
- private static final long RANDOM_SEED = System.currentTimeMillis();
- private static final Random RANDOM = new Random(RANDOM_SEED);
- static final LongMsgOffset PARTITION_OFFSET = new LongMsgOffset(RANDOM.nextInt(Integer.MAX_VALUE));
- static final long CURRENT_TIME_MS = System.currentTimeMillis();
- static final long START_TIME_MS = CURRENT_TIME_MS - TimeUnit.HOURS.toMillis(RANDOM.nextInt(24) + 24);
- static final long END_TIME_MS = START_TIME_MS + TimeUnit.HOURS.toMillis(RANDOM.nextInt(24) + 1);
- static final Interval INTERVAL = new Interval(START_TIME_MS, END_TIME_MS);
+ private static final String SEGMENT_LOCATION_PREFIX = "http://control_vip/segments/";
+ private static final int SEGMENT_SIZE_IN_BYTES = 100000000;
+
+ private static final long CURRENT_TIME_MS = System.currentTimeMillis();
+ private static final Random RANDOM = new Random(CURRENT_TIME_MS);
+ private static final LongMsgOffset PARTITION_OFFSET = new LongMsgOffset(RANDOM.nextInt(Integer.MAX_VALUE));
+ private static final int NUM_DOCS = RANDOM.nextInt(Integer.MAX_VALUE) + 1;
+ private static final String NEXT_OFFSET = new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString();
+ private static final long LATEST_OFFSET = PARTITION_OFFSET.getOffset() * 2 + NUM_DOCS;
+ private static final long START_TIME_MS = CURRENT_TIME_MS - TimeUnit.HOURS.toMillis(RANDOM.nextInt(24) + 24);
+ private static final long END_TIME_MS = START_TIME_MS + TimeUnit.HOURS.toMillis(RANDOM.nextInt(24) + 1);
+ private static final Interval INTERVAL = new Interval(START_TIME_MS, END_TIME_MS);
// NOTE: CRC is always non-negative
- static final String CRC = Long.toString(RANDOM.nextLong() & 0xFFFFFFFFL);
- static final String DATA_CRC = Long.toString(RANDOM.nextLong() & 0xFFFFFFFFL);
- static final SegmentVersion SEGMENT_VERSION = RANDOM.nextBoolean() ? SegmentVersion.v1 : SegmentVersion.v3;
- static final int NUM_DOCS = RANDOM.nextInt(Integer.MAX_VALUE) + 1;
- static final long LATEST_OFFSET = PARTITION_OFFSET.getOffset() * 2 + NUM_DOCS;
- static final int SEGMENT_SIZE_IN_BYTES = 100000000;
+ private static final String CRC = Long.toString(RANDOM.nextLong() & 0xFFFFFFFFL);
+ private static final String DATA_CRC = Long.toString(RANDOM.nextLong() & 0xFFFFFFFFL);
+ private static final SegmentVersion SEGMENT_VERSION = RANDOM.nextBoolean() ? SegmentVersion.v1 : SegmentVersion.v3;
@AfterClass
public void tearDown()
@@ -156,6 +157,22 @@ public class PinotLLCRealtimeSegmentManagerTest {
return segmentMetadata;
}
+ private CommittingSegmentDescriptor createCommittingSegmentDescriptor(String segmentName, String nextOffset,
+ String segmentLocation) {
+ CommittingSegmentDescriptor committingSegmentDescriptor =
+ new CommittingSegmentDescriptor(segmentName, nextOffset, SEGMENT_SIZE_IN_BYTES, segmentLocation);
+ committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ return committingSegmentDescriptor;
+ }
+
+ private CommittingSegmentDescriptor createCommittingSegmentDescriptor(String segmentName, String nextOffset) {
+ return createCommittingSegmentDescriptor(segmentName, nextOffset, SEGMENT_LOCATION_PREFIX + segmentName);
+ }
+
+ private CommittingSegmentDescriptor createCommittingSegmentDescriptor(String segmentName) {
+ return createCommittingSegmentDescriptor(segmentName, NEXT_OFFSET);
+ }
+
/**
* Test cases for new table being created, and initial segments setup that follows.
*/
@@ -226,8 +243,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
List<StreamMetadata> streamMetadataList = Collections.singletonList(
new StreamMetadata(segmentManager._streamConfigs.get(0), 3,
- Arrays.asList(
- new PartitionGroupMetadata(0, PARTITION_OFFSET, 5),
+ Arrays.asList(new PartitionGroupMetadata(0, PARTITION_OFFSET, 5),
new PartitionGroupMetadata(1, PARTITION_OFFSET, 10),
new PartitionGroupMetadata(2, PARTITION_OFFSET, 0))));
segmentManager.setUpNewTable(segmentManager._tableConfig, idealState, streamMetadataList);
@@ -264,8 +280,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
IdealState idealState = new IdealState(REALTIME_TABLE_NAME);
List<StreamMetadata> streamMetadataList = Collections.singletonList(
new StreamMetadata(segmentManager._streamConfigs.get(0), 2,
- Arrays.asList(
- new PartitionGroupMetadata(0, PARTITION_OFFSET),
+ Arrays.asList(new PartitionGroupMetadata(0, PARTITION_OFFSET),
new PartitionGroupMetadata(1, PARTITION_OFFSET))));
segmentManager.setUpNewTable(segmentManager._tableConfig, idealState, streamMetadataList);
@@ -301,10 +316,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Commit a segment for partition group 0
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- String nextOffset = new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString();
- CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(committingSegment, nextOffset, SEGMENT_SIZE_IN_BYTES);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
// Verify instance states for committed segment and new consuming segment
@@ -323,7 +335,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata committedSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(committingSegment);
assertEquals(committedSegmentZKMetadata.getStatus(), Status.DONE);
assertEquals(committedSegmentZKMetadata.getStartOffset(), PARTITION_OFFSET.toString());
- assertEquals(committedSegmentZKMetadata.getEndOffset(), nextOffset);
+ assertEquals(committedSegmentZKMetadata.getEndOffset(), NEXT_OFFSET);
assertEquals(committedSegmentZKMetadata.getCreationTime(), CURRENT_TIME_MS);
assertEquals(committedSegmentZKMetadata.getCrc(), Long.parseLong(CRC));
assertEquals(committedSegmentZKMetadata.getDataCrc(), Long.parseLong(DATA_CRC));
@@ -333,15 +345,14 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata consumingSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(consumingSegment);
assertEquals(consumingSegmentZKMetadata.getStatus(), Status.IN_PROGRESS);
- assertEquals(consumingSegmentZKMetadata.getStartOffset(), nextOffset);
+ assertEquals(consumingSegmentZKMetadata.getStartOffset(), NEXT_OFFSET);
assertEquals(committedSegmentZKMetadata.getCreationTime(), CURRENT_TIME_MS);
// Turn one instance of the consuming segment OFFLINE and commit the segment
consumingSegmentInstanceStateMap.entrySet().iterator().next().setValue(SegmentStateModel.OFFLINE);
committingSegment = consumingSegment;
- committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment,
+ new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS + NUM_DOCS).toString());
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
// Verify instance states for committed segment and new consuming segment
@@ -373,17 +384,14 @@ public class PinotLLCRealtimeSegmentManagerTest {
List<PartitionGroupMetadata> filteredList = new ArrayList<>(originalSm.getPartitionGroupMetadataList());
filteredList.remove(0);
segmentManager._streamMetadataList = Collections.singletonList(
- new StreamMetadata(originalSm.getStreamConfig(),
- originalSm.getNumPartitions(), filteredList));
+ new StreamMetadata(originalSm.getStreamConfig(), originalSm.getNumPartitions(), filteredList));
// Commit a segment for partition 0 - No new entries created for partition which reached end of life
committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 2, CURRENT_TIME_MS).getSegmentName();
String committingSegmentStartOffset = segmentManager._segmentZKMetadataMap.get(committingSegment).getStartOffset();
String committingSegmentEndOffset =
new LongMsgOffset(Long.parseLong(committingSegmentStartOffset) + NUM_DOCS).toString();
- committingSegmentDescriptor =
- new CommittingSegmentDescriptor(committingSegment, committingSegmentEndOffset, SEGMENT_SIZE_IN_BYTES);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment, committingSegmentEndOffset);
int instanceStateMapSize = instanceStatesMap.size();
int metadataMapSize = segmentManager._segmentZKMetadataMap.size();
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
@@ -431,8 +439,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Provide a segment that is not in CONSUMING state (non-existent); should be ignored without exception
String nonConsumingSegment = "nonExistingSegment";
- Set<String> committed = segmentManager.forceCommit(REALTIME_TABLE_NAME, null, nonConsumingSegment,
- BatchConfig.of(1, 1, 5));
+ Set<String> committed =
+ segmentManager.forceCommit(REALTIME_TABLE_NAME, null, nonConsumingSegment, BatchConfig.of(1, 1, 5));
assertTrue(committed.isEmpty(), "Expected no segments to be committed when only non-consuming segments provided");
}
@@ -452,8 +460,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
segmentManager._idealState.setPartitionState(consumingSegment, "Server_0", SegmentStateModel.CONSUMING);
try {
- Set<String> committed = segmentManager.forceCommit(REALTIME_TABLE_NAME, null, consumingSegment,
- BatchConfig.of(1, 1, 5));
+ Set<String> committed =
+ segmentManager.forceCommit(REALTIME_TABLE_NAME, null, consumingSegment, BatchConfig.of(1, 1, 5));
assertFalse(committed.isEmpty(), "Expected segments to be committed");
// If we reach here without exception, test passes
} catch (IllegalStateException e) {
@@ -495,8 +503,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
String consumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
pausedIdealState.setPartitionState(consumingSegment, "Server_0", SegmentStateModel.CONSUMING);
- doReturn(pausedIdealState).when(segmentManager).updatePauseStateInIdealState(eq(tableNameWithType), eq(true),
- eq(PauseState.ReasonCode.ADMINISTRATIVE), any());
+ doReturn(pausedIdealState).when(segmentManager)
+ .updatePauseStateInIdealState(eq(tableNameWithType), eq(true), eq(PauseState.ReasonCode.ADMINISTRATIVE), any());
BatchConfig batchConfig = BatchConfig.of(2, 3, 10);
final BatchConfig[] capturedConfig = new BatchConfig[1];
@@ -544,10 +552,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Commit a segment for partition group 0
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- String endOffset = new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString();
- CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(committingSegment, endOffset, SEGMENT_SIZE_IN_BYTES);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
// Verify instance states for committed segment and new consuming segment
@@ -566,7 +571,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata committedSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(committingSegment);
assertEquals(committedSegmentZKMetadata.getStatus(), Status.DONE);
assertEquals(committedSegmentZKMetadata.getStartOffset(), PARTITION_OFFSET.toString());
- assertEquals(committedSegmentZKMetadata.getEndOffset(), endOffset);
+ assertEquals(committedSegmentZKMetadata.getEndOffset(), NEXT_OFFSET);
assertEquals(committedSegmentZKMetadata.getCreationTime(), CURRENT_TIME_MS);
assertEquals(committedSegmentZKMetadata.getCrc(), Long.parseLong(CRC));
assertEquals(committedSegmentZKMetadata.getIndexVersion(), SEGMENT_VERSION.name());
@@ -611,10 +616,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Commit a segment for partition group 0
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- String endOffset = new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString();
- CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(committingSegment, endOffset, SEGMENT_SIZE_IN_BYTES);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
// Verify instance states for committed segment and new consuming segment
@@ -633,7 +635,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata committedSegmentZKMetadata = segmentManager._segmentZKMetadataMap.get(committingSegment);
assertEquals(committedSegmentZKMetadata.getStatus(), Status.DONE);
assertEquals(committedSegmentZKMetadata.getStartOffset(), PARTITION_OFFSET.toString());
- assertEquals(committedSegmentZKMetadata.getEndOffset(), endOffset);
+ assertEquals(committedSegmentZKMetadata.getEndOffset(), NEXT_OFFSET);
assertEquals(committedSegmentZKMetadata.getCreationTime(), CURRENT_TIME_MS);
assertEquals(committedSegmentZKMetadata.getCrc(), Long.parseLong(CRC));
assertEquals(committedSegmentZKMetadata.getIndexVersion(), SEGMENT_VERSION.name());
@@ -674,9 +676,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// 2 partitions commit segment
for (int partitionGroupId = 0; partitionGroupId < 2; partitionGroupId++) {
String segmentName = new LLCSegmentName(RAW_TABLE_NAME, partitionGroupId, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(segmentName,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(segmentName);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
}
testSetUpNewPartitions(segmentManager, false);
@@ -848,9 +848,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// 2 partitions commit segment
for (int partitionGroupId = 0; partitionGroupId < 2; partitionGroupId++) {
String segmentName = new LLCSegmentName(RAW_TABLE_NAME, partitionGroupId, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(segmentName,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(segmentName);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
}
@@ -891,9 +889,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
int sequenceNumber = partitionGroupId == 0 ? 2 : 0;
String segmentName =
new LLCSegmentName(RAW_TABLE_NAME, partitionGroupId, sequenceNumber, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(segmentName,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(segmentName);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
}
@@ -931,17 +927,15 @@ public class PinotLLCRealtimeSegmentManagerTest {
List<PartitionGroupMetadata> filteredPgList = new ArrayList<>(origSm.getPartitionGroupMetadataList());
filteredPgList.remove(1);
segmentManager._streamMetadataList = Collections.singletonList(
- new StreamMetadata(origSm.getStreamConfig(),
- origSm.getNumPartitions(), filteredPgList));
+ new StreamMetadata(origSm.getStreamConfig(), origSm.getNumPartitions(), filteredPgList));
// noop
testRepairs(segmentManager, Collections.emptyList());
// 1 commits segment - should not create new metadata or CONSUMING segment
String segmentName = new LLCSegmentName(RAW_TABLE_NAME, 1, 1, CURRENT_TIME_MS).getSegmentName();
String startOffset = segmentManager._segmentZKMetadataMap.get(segmentName).getStartOffset();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(segmentName,
- new LongMsgOffset(Long.parseLong(startOffset) + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(segmentName,
+ new LongMsgOffset(Long.parseLong(startOffset) + NUM_DOCS).toString());
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
// ONLINE in IS and metadata DONE, but end of shard (not present in partition group list), so don't repair
testRepairs(segmentManager, Lists.newArrayList(1));
@@ -975,10 +969,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
consumingSegmentInstanceStateMap.put(offlineInstance, SegmentStateModel.OFFLINE);
// Verify we have mixed state: 2 CONSUMING, 1 OFFLINE
- long consumingCount = consumingSegmentInstanceStateMap.values().stream()
- .filter(s -> s.equals(SegmentStateModel.CONSUMING)).count();
- long offlineCount = consumingSegmentInstanceStateMap.values().stream()
- .filter(s -> s.equals(SegmentStateModel.OFFLINE)).count();
+ long consumingCount =
+ consumingSegmentInstanceStateMap.values().stream().filter(s -> s.equals(SegmentStateModel.CONSUMING)).count();
+ long offlineCount =
+ consumingSegmentInstanceStateMap.values().stream().filter(s -> s.equals(SegmentStateModel.OFFLINE)).count();
assertEquals(consumingCount, 2);
assertEquals(offlineCount, 1);
@@ -1115,8 +1109,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
Map<String, String> instanceStateMap = entry.getValue();
// Skip segments with all instances OFFLINE
- if (instanceStateMap.containsValue(SegmentStateModel.ONLINE) || instanceStateMap
- .containsValue(SegmentStateModel.CONSUMING)) {
+ if (instanceStateMap.containsValue(SegmentStateModel.ONLINE) || instanceStateMap.containsValue(
+ SegmentStateModel.CONSUMING)) {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
int partitionsId = llcSegmentName.getPartitionGroupId();
Map<Integer, String> sequenceNumberToSegmentMap = partitionGroupIdToSegmentsMap.get(partitionsId);
@@ -1182,9 +1176,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Commit a segment for partition group 0
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
try {
segmentManager1.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
@@ -1287,8 +1279,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Expected
}
try {
- segmentManager.reduceSegmentSizeAndReset(new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS),
- 1000);
+ segmentManager.reduceSegmentSizeAndReset(new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS), 1000);
fail();
} catch (IllegalStateException e) {
// Expected
@@ -1312,22 +1303,18 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Test case 1: segment location with vip format.
// Commit a segment for partition group 0
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- String segmentLocationVIP = "http://control_vip/segments/segment1";
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, segmentLocationVIP);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
SegmentZKMetadata segmentZKMetadata =
segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, committingSegment, null);
- Assert.assertEquals(segmentZKMetadata.getDownloadUrl(), segmentLocationVIP);
+ Assert.assertEquals(segmentZKMetadata.getDownloadUrl(), SEGMENT_LOCATION_PREFIX + committingSegment);
- // Test case 2: segment location with peer format: peer://segment1, verify that an empty string is stored in zk.
+ // Test case 2: segment location with peer format: peer://segment, verify that an empty string is stored in zk.
committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
- String peerSegmentLocation = CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME + "/segment1";
- committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, peerSegmentLocation);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ String peerSegmentLocation = CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME + "/" + committingSegment;
+ committingSegmentDescriptor =
+ createCommittingSegmentDescriptor(committingSegment, NEXT_OFFSET, peerSegmentLocation);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
segmentZKMetadata = segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, committingSegment, null);
@@ -1342,9 +1329,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
setUpNewTable(segmentManager, 2, 5, 4);
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
@@ -1358,15 +1343,12 @@ public class PinotLLCRealtimeSegmentManagerTest {
spy(new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager));
setUpNewTable(segmentManager, 2, 5, 4);
segmentManager._streamMetadataList = Collections.singletonList(
- new StreamMetadata(segmentManager._streamConfigs.get(0), 4,
- IntStream.range(0, 4)
- .mapToObj(partition -> new PartitionGroupMetadata(partition, PARTITION_OFFSET))
- .collect(Collectors.toList())));
+ new StreamMetadata(segmentManager._streamConfigs.get(0), 4, IntStream.range(0, 4)
+ .mapToObj(partition -> new PartitionGroupMetadata(partition, PARTITION_OFFSET))
+ .collect(Collectors.toList())));
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
@@ -1378,29 +1360,24 @@ public class PinotLLCRealtimeSegmentManagerTest {
FakePinotLLCRealtimeSegmentManager segmentManager = spy(new FakePinotLLCRealtimeSegmentManager());
setUpNewTable(segmentManager, 2, 5, 4);
segmentManager._streamMetadataList = Collections.singletonList(
- new StreamMetadata(segmentManager._streamConfigs.get(0), 4,
- IntStream.range(0, 4)
- .mapToObj(partition -> new PartitionGroupMetadata(partition, PARTITION_OFFSET))
- .collect(Collectors.toList())));
+ new StreamMetadata(segmentManager._streamConfigs.get(0), 4, IntStream.range(0, 4)
+ .mapToObj(partition -> new PartitionGroupMetadata(partition, PARTITION_OFFSET))
+ .collect(Collectors.toList())));
- PauseState pauseState =
- new PauseState(false, PauseState.ReasonCode.ADMINISTRATIVE, "pause-topic-for-test",
- Long.toString(CURRENT_TIME_MS), Collections.singletonList(0));
- segmentManager._idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
- pauseState.toJsonString());
+ PauseState pauseState = new PauseState(false, PauseState.ReasonCode.ADMINISTRATIVE, "pause-topic-for-test",
+ Long.toString(CURRENT_TIME_MS), Collections.singletonList(0));
+ segmentManager._idealState.getRecord()
+ .setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE, pauseState.toJsonString());
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
String expectedNewConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
assertFalse(segmentManager._segmentZKMetadataMap.containsKey(expectedNewConsumingSegment));
assertFalse(segmentManager._idealState.getRecord().getMapFields().containsKey(expectedNewConsumingSegment));
- ZkHelixPropertyStore<ZNRecord> propertyStore =
- segmentManager._mockResourceManager.getPropertyStore();
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
verify(propertyStore, never()).remove(anyString(), eq(AccessOption.PERSISTENT));
}
@@ -1408,21 +1385,19 @@ public class PinotLLCRealtimeSegmentManagerTest {
public void testCommitSegmentMetadataCleansUpMetadataWhenTablePaused() {
FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
setUpNewTable(segmentManager, 2, 5, 4);
- ZkHelixPropertyStore<ZNRecord> propertyStore =
- segmentManager._mockResourceManager.getPropertyStore();
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
when(propertyStore.remove(anyString(), eq(AccessOption.PERSISTENT))).thenReturn(true);
- PauseState pauseState = new PauseState(true, PauseState.ReasonCode.ADMINISTRATIVE, "pause-for-test",
- Long.toString(CURRENT_TIME_MS), Collections.emptyList());
- segmentManager._idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
- pauseState.toJsonString());
+ PauseState pauseState =
+ new PauseState(true, PauseState.ReasonCode.ADMINISTRATIVE, "pause-for-test", Long.toString(CURRENT_TIME_MS),
+ Collections.emptyList());
+ segmentManager._idealState.getRecord()
+ .setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE, pauseState.toJsonString());
segmentManager._idealState.getRecord()
.setSimpleField(PinotLLCRealtimeSegmentManager.IS_TABLE_PAUSED, Boolean.TRUE.toString());
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
@@ -1436,20 +1411,16 @@ public class PinotLLCRealtimeSegmentManagerTest {
public void testCommitSegmentMetadataCleansUpMetadataWhenTopicPaused() {
FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
setUpNewTable(segmentManager, 2, 5, 4);
- ZkHelixPropertyStore<ZNRecord> propertyStore =
- segmentManager._mockResourceManager.getPropertyStore();
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
when(propertyStore.remove(anyString(), eq(AccessOption.PERSISTENT))).thenReturn(true);
- PauseState pauseState =
- new PauseState(false, PauseState.ReasonCode.ADMINISTRATIVE, "pause-topic-for-test",
- Long.toString(CURRENT_TIME_MS), Collections.singletonList(0));
- segmentManager._idealState.getRecord().setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE,
- pauseState.toJsonString());
+ PauseState pauseState = new PauseState(false, PauseState.ReasonCode.ADMINISTRATIVE, "pause-topic-for-test",
+ Long.toString(CURRENT_TIME_MS), Collections.singletonList(0));
+ segmentManager._idealState.getRecord()
+ .setSimpleField(PinotLLCRealtimeSegmentManager.PAUSE_STATE, pauseState.toJsonString());
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
String newConsumingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 1, CURRENT_TIME_MS).getSegmentName();
@@ -1463,17 +1434,16 @@ public class PinotLLCRealtimeSegmentManagerTest {
public void testCommitSegmentMetadataCleansUpMetadataWhenCommittingSegmentNotConsuming() {
FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
setUpNewTable(segmentManager, 2, 5, 4);
- ZkHelixPropertyStore<ZNRecord> propertyStore =
- segmentManager._mockResourceManager.getPropertyStore();
+ ZkHelixPropertyStore<ZNRecord> propertyStore = segmentManager._mockResourceManager.getPropertyStore();
when(propertyStore.remove(anyString(), eq(AccessOption.PERSISTENT))).thenReturn(true);
String committingSegment = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- segmentManager._idealState.getRecord().getMapFields().get(committingSegment)
+ segmentManager._idealState.getRecord()
+ .getMapFields()
+ .get(committingSegment)
.replaceAll((instance, state) -> SegmentStateModel.ONLINE);
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(committingSegment,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L, "http://control_vip/segments/1");
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(committingSegment);
try {
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
fail("Expected commitSegmentMetadata to fail when committing segment has no CONSUMING instance");
@@ -1892,11 +1862,13 @@ public class PinotLLCRealtimeSegmentManagerTest {
SegmentZKMetadata segZKMeta = mock(SegmentZKMetadata.class);
PinotHelixResourceManager helixResourceManager = mock(PinotHelixResourceManager.class);
- when(helixResourceManager.getTableConfig(REALTIME_TABLE_NAME))
- .thenReturn(new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setLLC(true)
- .setStreamConfigs(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap()).build());
- PinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager(
- helixResourceManager, config);
+ when(helixResourceManager.getTableConfig(REALTIME_TABLE_NAME)).thenReturn(
+ new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
+ .setLLC(true)
+ .setStreamConfigs(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap())
+ .build());
+ PinotLLCRealtimeSegmentManager segmentManager =
+ new FakePinotLLCRealtimeSegmentManager(helixResourceManager, config);
int numDeletedTmpSegments;
// case 1: the segmentMetadata download uri is identical to the uri of the tmp segment. Should not delete
@@ -1934,10 +1906,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
new PartitionGroupConsumptionStatus(1, 12, new LongMsgOffset(123), new LongMsgOffset(345), "ONLINE"));
doReturn(partitionGroupConsumptionStatusList).when(segmentManagerSpy)
.getPartitionGroupConsumptionStatusList(idealState, streamConfigs);
- List<StreamMetadata> streamMetadataList =
- List.of(new StreamMetadata(streamConfigs.get(0), 2,
- List.of(new PartitionGroupMetadata(0, new LongMsgOffset(234)),
- new PartitionGroupMetadata(1, new LongMsgOffset(345)))));
+ List<StreamMetadata> streamMetadataList = List.of(new StreamMetadata(streamConfigs.get(0), 2,
+ List.of(new PartitionGroupMetadata(0, new LongMsgOffset(234)),
+ new PartitionGroupMetadata(1, new LongMsgOffset(345)))));
doReturn(streamMetadataList).when(segmentManagerSpy)
.getNewStreamMetadataList(streamConfigs, partitionGroupConsumptionStatusList, idealState);
partitionIds = segmentManagerSpy.getPartitionIds(streamConfigs, idealState);
@@ -1959,9 +1930,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Commit segments for partitions 0 and 1 to get a mix of ONLINE (DONE) and CONSUMING (IN_PROGRESS) segments
for (int partitionGroupId = 0; partitionGroupId < 2; partitionGroupId++) {
String segmentName = new LLCSegmentName(RAW_TABLE_NAME, partitionGroupId, 0, CURRENT_TIME_MS).getSegmentName();
- CommittingSegmentDescriptor committingSegmentDescriptor = new CommittingSegmentDescriptor(segmentName,
- new LongMsgOffset(PARTITION_OFFSET.getOffset() + NUM_DOCS).toString(), 0L);
- committingSegmentDescriptor.setSegmentMetadata(mockSegmentMetadata());
+ CommittingSegmentDescriptor committingSegmentDescriptor = createCommittingSegmentDescriptor(segmentName);
segmentManager.commitSegmentMetadata(REALTIME_TABLE_NAME, committingSegmentDescriptor);
}
@@ -1970,12 +1939,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
for (Map.Entry<String, SegmentZKMetadata> entry : segmentManager._segmentZKMetadataMap.entrySet()) {
LLCSegmentName llcSegmentName = new LLCSegmentName(entry.getKey());
int partitionId = llcSegmentName.getPartitionGroupId();
- latestSegmentZKMetadataMap.merge(partitionId, entry.getValue(),
- (existing, candidate) -> {
- int existingSeq = new LLCSegmentName(existing.getSegmentName()).getSequenceNumber();
- int candidateSeq = new LLCSegmentName(candidate.getSegmentName()).getSequenceNumber();
- return candidateSeq > existingSeq ? candidate : existing;
- });
+ latestSegmentZKMetadataMap.merge(partitionId, entry.getValue(), (existing, candidate) -> {
+ int existingSeq = new LLCSegmentName(existing.getSegmentName()).getSequenceNumber();
+ int candidateSeq = new LLCSegmentName(candidate.getSegmentName()).getSequenceNumber();
+ return candidateSeq > existingSeq ? candidate : existing;
+ });
}
// Get results from both methods
@@ -2007,8 +1975,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
"Start offset mismatch for partition " + isStatus.getPartitionGroupId());
String zkEnd = zkStatus.getEndOffset() != null ? zkStatus.getEndOffset().toString() : null;
String isEnd = isStatus.getEndOffset() != null ? isStatus.getEndOffset().toString() : null;
- assertEquals(zkEnd, isEnd,
- "End offset mismatch for partition " + isStatus.getPartitionGroupId());
+ assertEquals(zkEnd, isEnd, "End offset mismatch for partition " + isStatus.getPartitionGroupId());
assertEquals(zkStatus.getStatus(), isStatus.getStatus(),
"Status mismatch for partition " + isStatus.getPartitionGroupId());
}
@@ -2028,8 +1995,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
segmentManager._tableConfig.setValidationConfig(segmentsValidationAndRetentionConfig);
String segmentName = new ArrayList<>(segmentManager._segmentZKMetadataMap.keySet()).get(0);
- SegmentZKMetadata segmentZKMetadata =
- segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentName, null);
+ SegmentZKMetadata segmentZKMetadata = segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentName, null);
int prevRowSize = segmentZKMetadata.getSizeThresholdToFlushSegment();
segmentManager.reduceSegmentSizeAndReset(new LLCSegmentName(segmentName), 100);
Assert.assertEquals(Math.min(100 / 2, prevRowSize / 2),
@@ -2042,13 +2008,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
FakePinotLLCRealtimeSegmentManager realtimeSegmentManager =
new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager);
IdealState idealState = mock(IdealState.class);
- Map<String, Map<String, String>> map = Map.of(
- "seg0", Map.of("i1", "CONSUMING", "i4", "ONLINE"),
- "seg1", Map.of("i2", "CONSUMING"),
- "seg2", Map.of("i3", "CONSUMING", "i2", "OFFLINE"),
- "seg3", Map.of("i4", "CONSUMING", "i2", "CONSUMING", "i3", "CONSUMING"),
- "seg4", Map.of("i5", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING")
- );
+ Map<String, Map<String, String>> map =
+ Map.of("seg0", Map.of("i1", "CONSUMING", "i4", "ONLINE"), "seg1", Map.of("i2", "CONSUMING"), "seg2",
+ Map.of("i3", "CONSUMING", "i2", "OFFLINE"), "seg3",
+ Map.of("i4", "CONSUMING", "i2", "CONSUMING", "i3", "CONSUMING"), "seg4",
+ Map.of("i5", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"));
ZNRecord znRecord = mock(ZNRecord.class);
when(znRecord.getMapFields()).thenReturn(map);
@@ -2058,13 +2022,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
Map<String, Queue<String>> instanceToConsumingSegments =
realtimeSegmentManager.getInstanceToConsumingSegments(idealState, targetConsumingSegment);
- assertEquals(instanceToConsumingSegments, Map.of(
- "i1", new LinkedList<>(List.of("seg0", "seg4")),
- "i2", new LinkedList<>(List.of("seg1", "seg3")),
- "i3", new LinkedList<>(List.of("seg2", "seg3", "seg4")),
- "i4", new LinkedList<>(List.of("seg3")),
- "i5", new LinkedList<>(List.of("seg4"))
- ));
+ assertEquals(instanceToConsumingSegments,
+ Map.of("i1", new LinkedList<>(List.of("seg0", "seg4")), "i2", new LinkedList<>(List.of("seg1", "seg3")), "i3",
+ new LinkedList<>(List.of("seg2", "seg3", "seg4")), "i4", new LinkedList<>(List.of("seg3")), "i5",
+ new LinkedList<>(List.of("seg4"))));
}
@Test
@@ -2074,15 +2035,13 @@ public class PinotLLCRealtimeSegmentManagerTest {
new FakePinotLLCRealtimeSegmentManager(mockHelixResourceManager);
IdealState idealState = mock(IdealState.class);
- Map<String, Map<String, String>> map = Map.of(
- "seg0", Map.of("i1", "CONSUMING", "i4", "ONLINE"),
- "seg1", Map.of("i2", "CONSUMING"),
- "seg2", Map.of("i3", "CONSUMING", "i2", "OFFLINE"),
- "seg3", Map.of("i4", "CONSUMING", "i2", "CONSUMING", "i3", "CONSUMING"),
- "seg4", Map.of("i5", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"),
- "seg5", Map.of("i6", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"),
- "seg6", Map.of("i7", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING")
- );
+ Map<String, Map<String, String>> map =
+ Map.of("seg0", Map.of("i1", "CONSUMING", "i4", "ONLINE"), "seg1", Map.of("i2", "CONSUMING"), "seg2",
+ Map.of("i3", "CONSUMING", "i2", "OFFLINE"), "seg3",
+ Map.of("i4", "CONSUMING", "i2", "CONSUMING", "i3", "CONSUMING"), "seg4",
+ Map.of("i5", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"), "seg5",
+ Map.of("i6", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"), "seg6",
+ Map.of("i7", "CONSUMING", "i1", "CONSUMING", "i3", "CONSUMING"));
ZNRecord znRecord = mock(ZNRecord.class);
when(znRecord.getMapFields()).thenReturn(map);
@@ -2092,18 +2051,11 @@ public class PinotLLCRealtimeSegmentManagerTest {
List<Set<String>> segmentBatchList =
realtimeSegmentManager.getSegmentBatchList(idealState, targetConsumingSegment, 2);
- assertEquals(segmentBatchList, List.of(
- Set.of("seg0", "seg1"),
- Set.of("seg2", "seg3"),
- Set.of("seg4", "seg5"),
- Set.of("seg6")
- ));
+ assertEquals(segmentBatchList,
+ List.of(Set.of("seg0", "seg1"), Set.of("seg2", "seg3"), Set.of("seg4", "seg5"), Set.of("seg6")));
segmentBatchList = realtimeSegmentManager.getSegmentBatchList(idealState, targetConsumingSegment, 4);
- assertEquals(segmentBatchList, List.of(
- Set.of("seg0", "seg1", "seg2", "seg3"),
- Set.of("seg4", "seg5", "seg6")
- ));
+ assertEquals(segmentBatchList, List.of(Set.of("seg0", "seg1", "seg2", "seg3"), Set.of("seg4", "seg5", "seg6")));
}
@Test
@@ -2158,12 +2110,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
String realtimeTableName = "githubEvents_2_REALTIME";
// Create test segments
- List<String> testSegments = List.of(
- "githubEvents_2__0__0__20250210T1142Z",
- "githubEvents_2__0__1__20250210T1142Z",
- "githubEvents_2__0__2__20250210T1142Z",
- "githubEvents_2__0__3__20250210T1142Z"
- );
+ List<String> testSegments = List.of("githubEvents_2__0__0__20250210T1142Z", "githubEvents_2__0__1__20250210T1142Z",
+ "githubEvents_2__0__2__20250210T1142Z", "githubEvents_2__0__3__20250210T1142Z");
// mock response of propertyStore
String committingSegmentsListPath =
@@ -2172,31 +2120,30 @@ public class PinotLLCRealtimeSegmentManagerTest {
ZNRecord znRecord = new ZNRecord(realtimeTableName);
znRecord.setListField(COMMITTING_SEGMENTS, testSegments);
- when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(znRecord);
+ when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(
+ znRecord);
// mock response for fetching segmentZKMetadata with different scenarios
// Segment 0: COMMITTING status
SegmentZKMetadata segmentZKMetadata0 = mock(SegmentZKMetadata.class);
when(segmentZKMetadata0.getStatus()).thenReturn(Status.COMMITTING);
- when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(0)))
- .thenReturn(segmentZKMetadata0);
+ when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(0))).thenReturn(
+ segmentZKMetadata0);
// Segment 1: null metadata (deleted)
- when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(1)))
- .thenReturn(null);
+ when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(1))).thenReturn(null);
// Segment 2: DONE status
SegmentZKMetadata segmentZKMetadata2 = mock(SegmentZKMetadata.class);
when(segmentZKMetadata2.getStatus()).thenReturn(Status.DONE);
- when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(2)))
- .thenReturn(segmentZKMetadata2);
+ when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(2))).thenReturn(
+ segmentZKMetadata2);
// Segment 3: COMMITTING status
SegmentZKMetadata segmentZKMetadata3 = mock(SegmentZKMetadata.class);
when(segmentZKMetadata3.getStatus()).thenReturn(Status.COMMITTING);
- when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(3)))
- .thenReturn(segmentZKMetadata3);
+ when(pinotHelixResourceManager.getSegmentZKMetadata(realtimeTableName, testSegments.get(3))).thenReturn(
+ segmentZKMetadata3);
// Execute test
List<String> result = segmentManager.getCommittingSegments(realtimeTableName);
@@ -2210,15 +2157,14 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertEquals(result, List.of(testSegments.get(3)));
// Test null case
- when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(null);
+ when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(null);
result = segmentManager.getCommittingSegments(realtimeTableName);
assertTrue(result.isEmpty());
// Test empty COMMITTING_SEGMENTS field
ZNRecord emptyRecord = new ZNRecord("CommittingSegments");
- when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(emptyRecord);
+ when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(
+ emptyRecord);
result = segmentManager.getCommittingSegments(realtimeTableName);
assertTrue(result.isEmpty());
}
@@ -2271,22 +2217,20 @@ public class PinotLLCRealtimeSegmentManagerTest {
// Test 1: Initial creation with mixed status segments
List<String> newSegments = Arrays.asList(committingSegment1, committingSegment2);
- when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(null);
- when(zkHelixPropertyStore.create(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(true);
+ when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(null);
+ when(zkHelixPropertyStore.create(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(
+ true);
assertTrue(segmentManager.syncCommittingSegments(realtimeTableName, newSegments));
// Test 2: Syncing with existing segments including DONE and missing metadata
ZNRecord existingRecord = new ZNRecord(realtimeTableName);
- existingRecord.setListField(COMMITTING_SEGMENTS,
- Arrays.asList(committingSegment2, doneSegment));
+ existingRecord.setListField(COMMITTING_SEGMENTS, Arrays.asList(committingSegment2, doneSegment));
- when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT)))
- .thenReturn(existingRecord);
- when(zkHelixPropertyStore.set(eq(committingSegmentsListPath), any(), anyInt(), eq(AccessOption.PERSISTENT)))
- .thenReturn(true);
+ when(zkHelixPropertyStore.get(eq(committingSegmentsListPath), any(), eq(AccessOption.PERSISTENT))).thenReturn(
+ existingRecord);
+ when(zkHelixPropertyStore.set(eq(committingSegmentsListPath), any(), anyInt(),
+ eq(AccessOption.PERSISTENT))).thenReturn(true);
// There should not be any duplicates and the doneSegment should be removed from the list
assertTrue(segmentManager.syncCommittingSegments(realtimeTableName,
@@ -2295,8 +2239,8 @@ public class PinotLLCRealtimeSegmentManagerTest {
new HashSet<>(List.of(committingSegment1, committingSegment2)));
// Test 3: Error handling during ZooKeeper operations
- when(zkHelixPropertyStore.set(eq(committingSegmentsListPath), any(), anyInt(), eq(AccessOption.PERSISTENT)))
- .thenThrow(new RuntimeException("ZooKeeper operation failed"));
+ when(zkHelixPropertyStore.set(eq(committingSegmentsListPath), any(), anyInt(),
+ eq(AccessOption.PERSISTENT))).thenThrow(new RuntimeException("ZooKeeper operation failed"));
assertFalse(segmentManager.syncCommittingSegments(realtimeTableName, newSegments));
}
@@ -2381,16 +2325,18 @@ public class PinotLLCRealtimeSegmentManagerTest {
void makeTableConfig() {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
- _tableConfig =
- new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(_numReplicas)
- .setStreamConfigs(streamConfigs).build();
+ _tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(_numReplicas)
+ .setStreamConfigs(streamConfigs)
+ .build();
_streamConfigs = IngestionConfigUtils.getStreamConfigs(_tableConfig);
}
void makeTableConfig(Map<String, String> streamConfigMap) {
- _tableConfig =
- new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(_numReplicas)
- .setStreamConfigs(streamConfigMap).build();
+ _tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME)
+ .setNumReplicas(_numReplicas)
+ .setStreamConfigs(streamConfigMap)
+ .build();
_streamConfigs = IngestionConfigUtils.getStreamConfigs(_tableConfig);
}
@@ -2476,9 +2422,10 @@ public class PinotLLCRealtimeSegmentManagerTest {
String newSegmentName, SegmentAssignment segmentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
Map<String, String> committingSegmentInstanceStateMap = _idealState.getInstanceStateMap(committingSegmentName);
- Preconditions.checkState(committingSegmentInstanceStateMap != null && committingSegmentInstanceStateMap
- .containsValue(SegmentStateModel.CONSUMING),
- "Failed to find instance in CONSUMING state in IdealState for segment: %s", committingSegmentName);
+ Preconditions.checkState(
+ committingSegmentInstanceStateMap != null && committingSegmentInstanceStateMap.containsValue(
+ SegmentStateModel.CONSUMING), "Failed to find instance in CONSUMING state in IdealState for segment: %s",
+ committingSegmentName);
updateInstanceStatesForNewConsumingSegment(_idealState.getRecord().getMapFields(), committingSegmentName,
isTablePaused(_idealState) || isTopicPaused(_idealState, committingSegmentName) ? null : newSegmentName,
segmentAssignment, instancePartitionsMap);
@@ -2499,9 +2446,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
if (_streamMetadataList != null) {
return _streamMetadataList;
} else {
- List<PartitionGroupMetadata> partitionGroupMetadataList =
- IntStream.range(0, _numPartitions).mapToObj(i -> new PartitionGroupMetadata(i, PARTITION_OFFSET))
- .collect(Collectors.toList());
+ List<PartitionGroupMetadata> partitionGroupMetadataList = IntStream.range(0, _numPartitions)
+ .mapToObj(i -> new PartitionGroupMetadata(i, PARTITION_OFFSET))
+ .collect(Collectors.toList());
return Collections.singletonList(
new StreamMetadata(streamConfigs.get(0), _numPartitions, partitionGroupMetadataList));
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java
index 7a9f01b2671..a85f22cd275 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdaterTest.java
@@ -139,8 +139,7 @@ public class FlushThresholdUpdaterTest {
for (long[] segmentSizesMB : Arrays.asList(EXPONENTIAL_GROWTH_SEGMENT_SIZES_MB, LOGARITHMIC_GROWTH_SEGMENT_SIZES_MB,
STEPS_SEGMENT_SIZES_MB)) {
- SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
- new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME, streamConfig.getTopicName());
+ SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -175,8 +174,7 @@ public class FlushThresholdUpdaterTest {
for (long[] segmentSizesMB : Arrays.asList(EXPONENTIAL_GROWTH_SEGMENT_SIZES_MB, LOGARITHMIC_GROWTH_SEGMENT_SIZES_MB,
STEPS_SEGMENT_SIZES_MB)) {
- SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
- new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME, streamConfig.getTopicName());
+ SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(1);
@@ -235,8 +233,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testTimeThreshold() {
StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
- SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
- new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME, streamConfig.getTopicName());
+ SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -267,8 +264,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testMinThreshold() {
StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
- SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
- new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME, streamConfig.getTopicName());
+ SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -304,8 +300,7 @@ public class FlushThresholdUpdaterTest {
int flushAutotuneInitialRows = StreamConfig.DEFAULT_FLUSH_AUTOTUNE_INITIAL_ROWS / 2;
StreamConfig streamConfig =
mockAutotuneStreamConfig(flushSegmentDesiredSizeBytes, flushThresholdTimeMillis, flushAutotuneInitialRows);
- SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater =
- new SegmentSizeBasedFlushThresholdUpdater(REALTIME_TABLE_NAME, streamConfig.getTopicName());
+ SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org