You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ja...@apache.org on 2023/09/16 01:50:21 UTC
[pinot] branch master updated: Remove/Deprecate HLC handling code (#11590)
This is an automated email from the ASF dual-hosted git repository.
jackie 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 e26a29ea14 Remove/Deprecate HLC handling code (#11590)
e26a29ea14 is described below
commit e26a29ea1491139149157d37c71f24635626d91b
Author: Xiaotian (Jackie) Jiang <17...@users.noreply.github.com>
AuthorDate: Fri Sep 15 18:50:15 2023 -0700
Remove/Deprecate HLC handling code (#11590)
---
...elixExternalViewBasedQueryQuotaManagerTest.java | 35 +-
.../pinot/common/metadata/ZKMetadataProvider.java | 17 +-
.../apache/pinot/common/utils/HLCSegmentName.java | 151 --------
.../apache/pinot/common/utils/LLCSegmentName.java | 81 ++--
.../org/apache/pinot/common/utils/SegmentName.java | 74 +---
.../pinot/common/utils/LLCSegmentNameTest.java | 98 +++++
.../pinot/common/utils/SegmentNameBuilderTest.java | 172 ---------
.../pinot/common/utils/config/TableConfigTest.java | 44 +--
.../common/utils/config/TableConfigUtilsTest.java | 21 +-
.../api/resources/PinotSegmentRestletResource.java | 5 +-
.../controller/helix/SegmentStatusChecker.java | 10 +-
.../helix/core/PinotHelixResourceManager.java | 247 ++++--------
.../helix/core/PinotTableIdealStateBuilder.java | 131 +------
.../helix/core/SegmentDeletionManager.java | 5 -
.../realtime/MissingConsumingSegmentFinder.java | 4 +-
.../realtime/PinotLLCRealtimeSegmentManager.java | 52 +--
.../core/realtime/PinotRealtimeSegmentManager.java | 421 ---------------------
.../core/realtime/SegmentCompletionManager.java | 8 +-
.../segment/DefaultFlushThresholdUpdater.java | 9 +-
.../segment/FlushThresholdUpdateManager.java | 8 +-
.../realtime/segment/FlushThresholdUpdater.java | 9 +-
.../segment/SegmentFlushThresholdComputer.java | 9 +-
.../SegmentSizeBasedFlushThresholdUpdater.java | 16 +-
.../helix/core/rebalance/TableRebalancer.java | 20 -
.../helix/core/relocation/SegmentRelocator.java | 14 +-
.../helix/core/retention/RetentionManager.java | 12 +-
.../helix/core/util/ZKMetadataUtils.java | 4 +-
.../RealtimeSegmentValidationManager.java | 56 +--
...anceAssignmentRestletResourceStatelessTest.java | 2 +-
.../api/TableConfigsRestletResourceTest.java | 3 +-
.../controller/api/upload/ZKOperatorTest.java | 2 +-
.../helix/RealtimeConsumerMonitorTest.java | 2 +-
.../controller/helix/SegmentStatusCheckerTest.java | 7 +-
...altimeNonReplicaGroupSegmentAssignmentTest.java | 4 +-
...NonReplicaGroupTieredSegmentAssignmentTest.java | 2 +-
.../RealtimeReplicaGroupSegmentAssignmentTest.java | 2 +-
.../SegmentAssignmentStrategyFactoryTest.java | 4 +-
.../PinotLLCRealtimeSegmentManagerTest.java | 36 +-
.../segment/FlushThresholdUpdaterTest.java | 25 +-
.../segment/SegmentFlushThresholdComputerTest.java | 68 ++--
.../helix/core/retention/RetentionManagerTest.java | 9 +-
.../validation/ValidationManagerTest.java | 23 +-
.../realtime/LLRealtimeSegmentDataManager.java | 33 +-
.../manager/realtime/RealtimeTableDataManager.java | 9 +-
.../realtime/RealtimeTableDataManagerTest.java | 7 -
.../realtime/SegmentCommitterFactoryTest.java | 5 +-
.../impl/fakestream/FakeStreamConfigUtils.java | 4 -
.../core/realtime/stream/StreamConfigTest.java | 409 +++++++-------------
.../tests/BaseClusterIntegrationTest.java | 72 +---
.../MergeRollupMinionClusterIntegrationTest.java | 6 +-
.../tests/RealtimeKinesisIntegrationTest.java | 7 +-
...fflineSegmentsMinionClusterIntegrationTest.java | 2 +-
.../UpsertTableSegmentPreloadIntegrationTest.java | 10 +-
.../UpsertTableSegmentUploadIntegrationTest.java | 2 +-
.../RealtimeToOfflineSegmentsTaskGenerator.java | 8 -
...RealtimeToOfflineSegmentsTaskGeneratorTest.java | 4 -
.../KafkaPartitionLevelStreamConfigTest.java | 4 -
.../plugin/stream/kinesis/KinesisConsumerTest.java | 2 -
.../RealtimeIndexOffHeapMemoryManager.java | 5 -
...oncurrentMapPartitionUpsertMetadataManager.java | 4 +-
.../segment/local/utils/ReplicationUtils.java | 62 ---
.../segment/local/utils/TableConfigUtils.java | 51 +--
.../pinot/server/api/resources/TablesResource.java | 2 +-
.../SegmentsValidationAndRetentionConfig.java | 39 +-
.../apache/pinot/spi/config/table/TableConfig.java | 26 +-
.../spi/stream/PartitionLevelStreamConfig.java | 51 +--
.../org/apache/pinot/spi/stream/StreamConfig.java | 141 +++----
.../pinot/spi/stream/StreamConfigProperties.java | 1 +
.../spi/utils/builder/TableConfigBuilder.java | 7 +-
.../apache/pinot/spi/config/ConfigUtilsTest.java | 5 -
70 files changed, 680 insertions(+), 2220 deletions(-)
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/queryquota/HelixExternalViewBasedQueryQuotaManagerTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/queryquota/HelixExternalViewBasedQueryQuotaManagerTest.java
index 80021f691c..64b2693767 100644
--- a/pinot-broker/src/test/java/org/apache/pinot/broker/queryquota/HelixExternalViewBasedQueryQuotaManagerTest.java
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/queryquota/HelixExternalViewBasedQueryQuotaManagerTest.java
@@ -147,8 +147,7 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
throws Exception {
ExternalView brokerResource = generateBrokerResource(OFFLINE_TABLE_NAME);
TableConfig tableConfig = generateDefaultTableConfig(OFFLINE_TABLE_NAME);
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
setQps(tableConfig);
_queryQuotaManager.initOrUpdateTableQueryQuota(tableConfig, brokerResource);
Assert.assertEquals(_queryQuotaManager.getRateLimiterMapSize(), 1);
@@ -179,7 +178,7 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setQuotaConfig(quotaConfig)
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("1").setSegmentPushType("APPEND")
.setBrokerTenant("testBroker").setServerTenant("testServer").build();
- ZKMetadataProvider.setRealtimeTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
TableConfigUtils.toZNRecord(realtimeTableConfig));
ExternalView brokerResource = generateBrokerResource(OFFLINE_TABLE_NAME);
@@ -202,7 +201,7 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setQuotaConfig(quotaConfig)
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("1").setSegmentPushType("APPEND")
.setBrokerTenant("testBroker").setServerTenant("testServer").build();
- ZKMetadataProvider.setRealtimeTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
TableConfigUtils.toZNRecord(realtimeTableConfig));
ExternalView brokerResource = generateBrokerResource(REALTIME_TABLE_NAME);
@@ -234,10 +233,10 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("1").setSegmentPushType("APPEND")
.setBrokerTenant("testBroker").setServerTenant("testServer").build();
- ZKMetadataProvider.setRealtimeTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
TableConfigUtils.toZNRecord(realtimeTableConfig));
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(offlineTableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME,
+ TableConfigUtils.toZNRecord(offlineTableConfig));
// Since each table has 2 online brokers, per broker rate becomes 100.0 / 2 = 50.0
_queryQuotaManager.initOrUpdateTableQueryQuota(offlineTableConfig, brokerResource);
@@ -263,8 +262,8 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
throws Exception {
ExternalView brokerResource = generateBrokerResource(REALTIME_TABLE_NAME);
TableConfig tableConfig = generateDefaultTableConfig(REALTIME_TABLE_NAME);
- ZKMetadataProvider
- .setRealtimeTableConfig(_testPropertyStore, REALTIME_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
+ TableConfigUtils.toZNRecord(tableConfig));
setQps(tableConfig);
_queryQuotaManager.initOrUpdateTableQueryQuota(tableConfig, brokerResource);
Assert.assertEquals(_queryQuotaManager.getRateLimiterMapSize(), 1);
@@ -280,8 +279,8 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
throws Exception {
ExternalView brokerResource = generateBrokerResource(REALTIME_TABLE_NAME);
TableConfig tableConfig = generateDefaultTableConfig(REALTIME_TABLE_NAME);
- ZKMetadataProvider
- .setRealtimeTableConfig(_testPropertyStore, REALTIME_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, REALTIME_TABLE_NAME,
+ TableConfigUtils.toZNRecord(tableConfig));
setQps(tableConfig);
_queryQuotaManager.initOrUpdateTableQueryQuota(tableConfig, brokerResource);
Assert.assertEquals(_queryQuotaManager.getRateLimiterMapSize(), 1);
@@ -312,8 +311,8 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setQuotaConfig(quotaConfig)
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("1").setSegmentPushType("APPEND")
.setBrokerTenant("testBroker").setServerTenant("testServer").build();
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(offlineTableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME,
+ TableConfigUtils.toZNRecord(offlineTableConfig));
ExternalView brokerResource = generateBrokerResource(REALTIME_TABLE_NAME);
TableConfig tableConfig = generateDefaultTableConfig(REALTIME_TABLE_NAME);
@@ -331,8 +330,8 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setQuotaConfig(quotaConfig)
.setRetentionTimeUnit("DAYS").setRetentionTimeValue("1").setSegmentPushType("APPEND")
.setBrokerTenant("testBroker").setServerTenant("testServer").build();
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(offlineTableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME,
+ TableConfigUtils.toZNRecord(offlineTableConfig));
ExternalView brokerResource = generateBrokerResource(OFFLINE_TABLE_NAME);
TableConfig tableConfig = generateDefaultTableConfig(REALTIME_TABLE_NAME);
@@ -358,8 +357,7 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
throws Exception {
ExternalView brokerResource = new ExternalView(CommonConstants.Helix.BROKER_RESOURCE_INSTANCE);
TableConfig tableConfig = generateDefaultTableConfig(OFFLINE_TABLE_NAME);
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
setQps(tableConfig);
_queryQuotaManager.initOrUpdateTableQueryQuota(tableConfig, brokerResource);
Assert.assertEquals(_queryQuotaManager.getRateLimiterMapSize(), 1);
@@ -371,8 +369,7 @@ public class HelixExternalViewBasedQueryQuotaManagerTest {
ExternalView brokerResource = new ExternalView(CommonConstants.Helix.BROKER_RESOURCE_INSTANCE);
brokerResource.setState(OFFLINE_TABLE_NAME, "broker_instance_2", "OFFLINE");
TableConfig tableConfig = generateDefaultTableConfig(OFFLINE_TABLE_NAME);
- ZKMetadataProvider
- .setOfflineTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
+ ZKMetadataProvider.setTableConfig(_testPropertyStore, OFFLINE_TABLE_NAME, TableConfigUtils.toZNRecord(tableConfig));
setQps(tableConfig);
_queryQuotaManager.initOrUpdateTableQueryQuota(tableConfig, brokerResource);
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
index c6425ca0e0..fc76f96709 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java
@@ -34,8 +34,8 @@ import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.SchemaUtils;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.config.AccessControlUserConfigUtils;
import org.apache.pinot.common.utils.config.TableConfigUtils;
import org.apache.pinot.spi.config.ConfigUtils;
@@ -73,15 +73,22 @@ public class ZKMetadataProvider {
propertyStore.set(constructPropertyStorePathForUserConfig(username), znRecord, AccessOption.PERSISTENT);
}
- public static void setRealtimeTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String realtimeTableName,
+ public static void setTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableNameWithType,
ZNRecord znRecord) {
- propertyStore.set(constructPropertyStorePathForResourceConfig(realtimeTableName), znRecord,
+ propertyStore.set(constructPropertyStorePathForResourceConfig(tableNameWithType), znRecord,
AccessOption.PERSISTENT);
}
+ @Deprecated
+ public static void setRealtimeTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String realtimeTableName,
+ ZNRecord znRecord) {
+ setTableConfig(propertyStore, realtimeTableName, znRecord);
+ }
+
+ @Deprecated
public static void setOfflineTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String offlineTableName,
ZNRecord znRecord) {
- propertyStore.set(constructPropertyStorePathForResourceConfig(offlineTableName), znRecord, AccessOption.PERSISTENT);
+ setTableConfig(propertyStore, offlineTableName, znRecord);
}
public static void setInstanceZKMetadata(ZkHelixPropertyStore<ZNRecord> propertyStore,
@@ -503,7 +510,7 @@ public class ZKMetadataProvider {
if (propertyStore.exists(segmentsPath, AccessOption.PERSISTENT)) {
List<String> segments = propertyStore.getChildNames(segmentsPath, AccessOption.PERSISTENT);
for (String segment : segments) {
- if (SegmentName.isLowLevelConsumerSegmentName(segment)) {
+ if (LLCSegmentName.isLLCSegment(segment)) {
llcRealtimeSegments.add(segment);
}
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/HLCSegmentName.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/HLCSegmentName.java
deleted file mode 100644
index e3b70af8d6..0000000000
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/HLCSegmentName.java
+++ /dev/null
@@ -1,151 +0,0 @@
-/**
- * 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.common.utils;
-
-// old style name (that needs to be recognized, but not built anymore)
-// return StringUtils.join(
-// Lists.newArrayList(tableName, instanceName, groupId, partitionName, sequenceNumber, creationTime), "__");
-// Example tableName_REALTIME__Server_host_name.domain.com_8001__tableName_REALTIME_1449605490946_1__0__1450910568438
-// Splitting this:
-// table name: tableName_REALTIME,
-// instanceName: Server_host_name.domain.com_8001
-// groupId: tableName_REALTIME_1449605490946_1 (has tablename, group creation time, and a group sequence nuumber)
-// sequenceNumber: 0
-// creationTime: 1450910568438
-//
-// shorter name: {groupId}__{partitionRange}__{sequenceNumber}
-// Example: tableName_REALTIME_1433316466991_0__0__1465314044238
-// Splitting this:
-// groupId: tableName_REALTIME_1433316466991_0
-// partitionRange: 0
-// sequenceNumber: 1465314044238
-// see org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder#getGroupIdFromRealtimeDataTable for details
-// on the groupId
-
-import com.google.common.collect.Lists;
-import org.apache.commons.lang.StringUtils;
-
-
-public class HLCSegmentName extends SegmentName {
- private final boolean _isOldStyleNaming;
- private final String _groupId;
- private final String _sequenceNumber;
- private final String _partitionRange;
- private final String _segmentName;
- private final RealtimeSegmentType _segmentType;
- private final String _tableName;
-
- // Can be called with old or new style naming.
- public HLCSegmentName(String segmentName) {
- // Decide if it is old style or new style v1 naming here.
- if (!isHighLevelConsumerSegmentName(segmentName)) {
- throw new RuntimeException(segmentName + " is not a HighLevelConsumer segment name");
- }
- String[] parts = segmentName.split(SEPARATOR);
- if (parts.length == 5) {
- _isOldStyleNaming = true;
- } else if (parts.length == 3) {
- _isOldStyleNaming = false;
- } else {
- throw new RuntimeException(segmentName + " is not HighLevelConsumer segment name ");
- }
-
- if (isOldStyleNaming()) {
- _tableName = parts[0];
- _groupId = parts[2];
- _partitionRange = parts[3];
- _sequenceNumber = parts[4];
- _segmentType = RealtimeSegmentType.HLC_LONG;
- } else {
- _groupId = parts[0];
- _partitionRange = parts[1];
- _sequenceNumber = parts[2];
- _tableName = _groupId.substring(0, _groupId.indexOf(REALTIME_SUFFIX) + REALTIME_SUFFIX_LENGTH);
- _segmentType = RealtimeSegmentType.HLC_SHORT;
- }
- _segmentName = segmentName;
- }
-
- /**
- * Builds only the newer style.
- * @param groupId is like myTable_REALTIME_1442428556382_0 (where 1442428556382 is the time when the group was created
- * or it is line myTable
- * @param partitionRange
- * @param sequenceNumber
- */
- public HLCSegmentName(String groupId, String partitionRange, String sequenceNumber) {
- if (isValidComponentName(groupId) && isValidComponentName(partitionRange) && isValidComponentName(sequenceNumber)) {
- _isOldStyleNaming = false;
- _groupId = groupId;
- _partitionRange = partitionRange;
- _sequenceNumber = sequenceNumber;
- _segmentName = StringUtils.join(Lists.newArrayList(groupId, partitionRange, sequenceNumber), SEPARATOR);
- _tableName = _groupId.substring(0, _groupId.indexOf(REALTIME_SUFFIX) + REALTIME_SUFFIX_LENGTH);
- _segmentType = RealtimeSegmentType.HLC_SHORT;
- } else {
- throw new IllegalArgumentException(
- "Invalid group id (" + groupId + "), partition range (" + partitionRange + ") or sequence number ("
- + sequenceNumber + ")");
- }
- }
-
- public boolean isOldStyleNaming() {
- return _isOldStyleNaming;
- }
-
- @Override
- public String getGroupId() {
- return _groupId;
- }
-
- @Override
- public String getSequenceNumberStr() {
- return _sequenceNumber;
- }
-
- @Override
- public int getSequenceNumber() {
- return Integer.valueOf(_sequenceNumber);
- }
-
- @Override
- public String getPartitionRange() {
- return _partitionRange;
- }
-
- @Override
- public String getSegmentName() {
- return _segmentName;
- }
-
- @Override
- public String getTableName() {
- return _tableName;
- }
-
- @Override
- public RealtimeSegmentType getSegmentType() {
- return _segmentType;
- }
-
- @Override
- public String toString() {
- return getSegmentName();
- }
-}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java
index 2a533a0e68..d72ed76427 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/LLCSegmentName.java
@@ -27,9 +27,10 @@ import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
-public class LLCSegmentName extends SegmentName implements Comparable {
- private final static String DATE_FORMAT = "yyyyMMdd'T'HHmm'Z'";
- private final static DateTimeFormatter DATE_FORMATTER = DateTimeFormat.forPattern(DATE_FORMAT).withZoneUTC();
+public class LLCSegmentName implements Comparable<LLCSegmentName> {
+ private static final String SEPARATOR = "__";
+ private static final String DATE_FORMAT = "yyyyMMdd'T'HHmm'Z'";
+ private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormat.forPattern(DATE_FORMAT).withZoneUTC();
private final String _tableName;
private final int _partitionGroupId;
@@ -48,9 +49,7 @@ public class LLCSegmentName extends SegmentName implements Comparable {
}
public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
- if (!isValidComponentName(tableName)) {
- throw new RuntimeException("Invalid table name " + tableName);
- }
+ Preconditions.checkArgument(!tableName.contains(SEPARATOR), "Illegal table name: %s", tableName);
_tableName = tableName;
_partitionGroupId = partitionGroupId;
_sequenceNumber = sequenceNumber;
@@ -81,6 +80,24 @@ public class LLCSegmentName extends SegmentName implements Comparable {
return new LLCSegmentName(parts[0], Integer.parseInt(parts[1]), Integer.parseInt(parts[2]), parts[3], segmentName);
}
+ /**
+ * Returns whether the given segment name represents an LLC segment.
+ */
+ public static boolean isLLCSegment(String segmentName) {
+ int numSeparators = 0;
+ int index = 0;
+ while ((index = segmentName.indexOf(SEPARATOR, index)) != -1) {
+ numSeparators++;
+ index += 2; // SEPARATOR.length()
+ }
+ return numSeparators == 3;
+ }
+
+ @Deprecated
+ public static boolean isLowLevelConsumerSegmentName(String segmentName) {
+ return isLLCSegment(segmentName);
+ }
+
/**
* Returns the sequence number of the given segment name.
*/
@@ -88,27 +105,14 @@ public class LLCSegmentName extends SegmentName implements Comparable {
return Integer.parseInt(StringUtils.splitByWholeSeparator(segmentName, SEPARATOR)[2]);
}
- @Override
- public RealtimeSegmentType getSegmentType() {
- return RealtimeSegmentType.LLC;
- }
-
- @Override
public String getTableName() {
return _tableName;
}
- @Override
public int getPartitionGroupId() {
return _partitionGroupId;
}
- @Override
- public String getPartitionRange() {
- return Integer.toString(getPartitionGroupId());
- }
-
- @Override
public int getSequenceNumber() {
return _sequenceNumber;
}
@@ -122,41 +126,18 @@ public class LLCSegmentName extends SegmentName implements Comparable {
return dateTime.getMillis();
}
- @Override
public String getSegmentName() {
return _segmentName;
}
@Override
- public String getSequenceNumberStr() {
- return Integer.toString(_sequenceNumber);
- }
-
- @Override
- public int compareTo(Object o) {
- LLCSegmentName other = (LLCSegmentName) o;
- if (!this.getTableName().equals(other.getTableName())) {
- throw new RuntimeException(
- "Cannot compare segment names " + this.getSegmentName() + " and " + other.getSegmentName());
- }
- if (this.getPartitionGroupId() > other.getPartitionGroupId()) {
- return 1;
- } else if (this.getPartitionGroupId() < other.getPartitionGroupId()) {
- return -1;
- } else {
- if (this.getSequenceNumber() > other.getSequenceNumber()) {
- return 1;
- } else if (this.getSequenceNumber() < other.getSequenceNumber()) {
- return -1;
- } else {
- if (!this.getCreationTime().equals(other.getCreationTime())) {
- // If sequence number is the same, time cannot be different.
- throw new RuntimeException(
- "Cannot compare segment names " + this.getSegmentName() + " and " + other.getSegmentName());
- }
- return 0;
- }
+ public int compareTo(LLCSegmentName other) {
+ Preconditions.checkArgument(_tableName.equals(other._tableName),
+ "Cannot compare segment names from different table: %s, %s", _segmentName, other.getSegmentName());
+ if (_partitionGroupId != other._partitionGroupId) {
+ return Integer.compare(_partitionGroupId, other._partitionGroupId);
}
+ return Integer.compare(_sequenceNumber, other._sequenceNumber);
}
@Override
@@ -164,7 +145,7 @@ public class LLCSegmentName extends SegmentName implements Comparable {
if (this == o) {
return true;
}
- if (o == null || getClass() != o.getClass()) {
+ if (!(o instanceof LLCSegmentName)) {
return false;
}
LLCSegmentName that = (LLCSegmentName) o;
@@ -178,6 +159,6 @@ public class LLCSegmentName extends SegmentName implements Comparable {
@Override
public String toString() {
- return getSegmentName();
+ return _segmentName;
}
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/SegmentName.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/SegmentName.java
index 503177ca98..2395e18e4e 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/SegmentName.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/SegmentName.java
@@ -18,80 +18,16 @@
*/
package org.apache.pinot.common.utils;
-public abstract class SegmentName {
- public static final String SEPARATOR = "__";
- public static final String REALTIME_SUFFIX = "_REALTIME";
- public static final int REALTIME_SUFFIX_LENGTH = REALTIME_SUFFIX.length();
-
- public enum RealtimeSegmentType {
- UNSUPPORTED, HLC_LONG, HLC_SHORT, LLC,
- }
-
- public static RealtimeSegmentType getSegmentType(String segmentName) {
- if (isHighLevelConsumerSegmentName(segmentName)) {
- HLCSegmentName segName = new HLCSegmentName(segmentName);
-
- if (segName.isOldStyleNaming()) {
- return RealtimeSegmentType.HLC_LONG;
- } else {
- return RealtimeSegmentType.HLC_SHORT;
- }
- }
-
- if (isLowLevelConsumerSegmentName(segmentName)) {
- return RealtimeSegmentType.LLC;
- }
-
- return RealtimeSegmentType.UNSUPPORTED;
- }
-
- protected boolean isValidComponentName(String string) {
- return !string.contains(SEPARATOR);
- }
-
- public abstract String getTableName();
-
- public abstract String getSequenceNumberStr();
-
- public abstract int getSequenceNumber();
-
- public abstract String getSegmentName();
-
- public abstract RealtimeSegmentType getSegmentType();
-
- public String getGroupId() {
- throw new RuntimeException("No groupId in " + getSegmentName());
- }
-
- public int getPartitionGroupId() {
- throw new RuntimeException("No partitionGroupId in " + getSegmentName());
- }
-
- public String getPartitionRange() {
- throw new RuntimeException("No partitionRange in " + getSegmentName());
- }
-
- public static boolean isHighLevelConsumerSegmentName(String segmentName) {
- int numSeparators = getNumSeparators(segmentName);
- return numSeparators == 2 || numSeparators == 4;
+@Deprecated
+public class SegmentName {
+ private SegmentName() {
}
public static boolean isLowLevelConsumerSegmentName(String segmentName) {
- return getNumSeparators(segmentName) == 3;
+ return LLCSegmentName.isLLCSegment(segmentName);
}
public static boolean isRealtimeSegmentName(String segmentName) {
- int numSeparators = getNumSeparators(segmentName);
- return numSeparators >= 2 && numSeparators <= 4;
- }
-
- private static int getNumSeparators(String segmentName) {
- int numSeparators = 0;
- int index = 0;
- while ((index = segmentName.indexOf(SEPARATOR, index)) != -1) {
- numSeparators++;
- index += 2; // SEPARATOR.length()
- }
- return numSeparators;
+ return LLCSegmentName.isLLCSegment(segmentName);
}
}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/LLCSegmentNameTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/LLCSegmentNameTest.java
new file mode 100644
index 0000000000..04b2a9e2fb
--- /dev/null
+++ b/pinot-common/src/test/java/org/apache/pinot/common/utils/LLCSegmentNameTest.java
@@ -0,0 +1,98 @@
+/**
+ * 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.common.utils;
+
+import java.util.Arrays;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/**
+ * Tests for the realtime segment name builder.
+ */
+public class LLCSegmentNameTest {
+
+ @Test
+ public void testSegmentNameBuilder() {
+ LLCSegmentName llcSegmentName = new LLCSegmentName("myTable", 0, 1, 1465508537069L);
+ String segmentName = llcSegmentName.getSegmentName();
+ assertEquals(segmentName, "myTable__0__1__20160609T2142Z");
+ assertTrue(LLCSegmentName.isLLCSegment(segmentName));
+ assertEquals(llcSegmentName.getTableName(), "myTable");
+ assertEquals(llcSegmentName.getPartitionGroupId(), 0);
+ assertEquals(llcSegmentName.getSequenceNumber(), 1);
+
+ // Invalid segment name
+ assertFalse(LLCSegmentName.isLLCSegment("a__abc__1__3__4__54__g__gg___h"));
+ }
+
+ @Test
+ public void testLLCSegmentName() {
+ String tableName = "myTable";
+ final int partitionGroupId = 4;
+ final int sequenceNumber = 27;
+ final long msSinceEpoch = 1466200248000L;
+ final String creationTime = "20160617T2150Z";
+ final long creationTimeInMs = 1466200200000L;
+ final String segmentName = "myTable__4__27__" + creationTime;
+
+ LLCSegmentName segName1 = new LLCSegmentName(tableName, partitionGroupId, sequenceNumber, msSinceEpoch);
+ Assert.assertEquals(segName1.getSegmentName(), segmentName);
+ Assert.assertEquals(segName1.getPartitionGroupId(), partitionGroupId);
+ Assert.assertEquals(segName1.getCreationTime(), creationTime);
+ Assert.assertEquals(segName1.getCreationTimeMs(), creationTimeInMs);
+ Assert.assertEquals(segName1.getSequenceNumber(), sequenceNumber);
+ Assert.assertEquals(segName1.getTableName(), tableName);
+
+ LLCSegmentName segName2 = new LLCSegmentName(segmentName);
+ Assert.assertEquals(segName2.getSegmentName(), segmentName);
+ Assert.assertEquals(segName2.getPartitionGroupId(), partitionGroupId);
+ Assert.assertEquals(segName2.getCreationTime(), creationTime);
+ Assert.assertEquals(segName2.getCreationTimeMs(), creationTimeInMs);
+ Assert.assertEquals(segName2.getSequenceNumber(), sequenceNumber);
+ Assert.assertEquals(segName2.getTableName(), tableName);
+
+ Assert.assertEquals(segName1, segName2);
+
+ LLCSegmentName segName3 = new LLCSegmentName(tableName, partitionGroupId + 1, sequenceNumber - 1, msSinceEpoch);
+ Assert.assertTrue(segName1.compareTo(segName3) < 0);
+ LLCSegmentName segName4 = new LLCSegmentName(tableName, partitionGroupId + 1, sequenceNumber + 1, msSinceEpoch);
+ Assert.assertTrue(segName1.compareTo(segName4) < 0);
+ LLCSegmentName segName5 = new LLCSegmentName(tableName, partitionGroupId - 1, sequenceNumber + 1, msSinceEpoch);
+ Assert.assertTrue(segName1.compareTo(segName5) > 0);
+ LLCSegmentName segName6 = new LLCSegmentName(tableName, partitionGroupId, sequenceNumber + 1, msSinceEpoch);
+ Assert.assertTrue(segName1.compareTo(segName6) < 0);
+
+ LLCSegmentName segName7 =
+ new LLCSegmentName(tableName + "NotGood", partitionGroupId, sequenceNumber + 1, msSinceEpoch);
+ try {
+ segName1.compareTo(segName7);
+ Assert.fail("Not failing when comparing " + segName1.getSegmentName() + " and " + segName7.getSegmentName());
+ } catch (Exception e) {
+ // expected
+ }
+ LLCSegmentName[] testSorted = new LLCSegmentName[]{segName3, segName1, segName4, segName5, segName6};
+ Arrays.sort(testSorted);
+ Assert.assertEquals(testSorted, new LLCSegmentName[]{segName5, segName1, segName6, segName3, segName4});
+ }
+}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/SegmentNameBuilderTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/SegmentNameBuilderTest.java
deleted file mode 100644
index 0fc15bc39f..0000000000
--- a/pinot-common/src/test/java/org/apache/pinot/common/utils/SegmentNameBuilderTest.java
+++ /dev/null
@@ -1,172 +0,0 @@
-/**
- * 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.common.utils;
-
-import java.util.Arrays;
-import org.testng.Assert;
-import org.testng.annotations.Test;
-
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.fail;
-
-
-/**
- * Tests for the realtime segment name builder.
- */
-public class SegmentNameBuilderTest {
- @Test
- public void testSegmentNameBuilder() {
- final String oldV1Name = "myTable_REALTIME__Server_1.2.3.4_1234__myTable_REALTIME_1234567_0__0__23456789";
- final String groupId = "myTable_REALTIME_1234567_0";
- final String partitionRange = "ALL";
- final String sequenceNumber = "1234567";
- HLCSegmentName shortNameSegment = new HLCSegmentName(groupId, partitionRange, sequenceNumber);
- HLCSegmentName longNameSegment = new HLCSegmentName(oldV1Name);
- final String shortV1Name = shortNameSegment.getSegmentName();
- LLCSegmentName llcSegment = new LLCSegmentName("myTable", 0, 1, 1465508537069L);
- final String v2Name = llcSegment.getSegmentName();
-
- assertEquals(shortV1Name, "myTable_REALTIME_1234567_0__ALL__1234567");
- assertEquals(v2Name, "myTable__0__1__20160609T2142Z");
-
- // Check v1/v2 format detection
- assertEquals(SegmentName.getSegmentType(oldV1Name), SegmentName.RealtimeSegmentType.HLC_LONG);
- assertEquals(SegmentName.getSegmentType(shortV1Name), SegmentName.RealtimeSegmentType.HLC_SHORT);
- assertEquals(SegmentName.getSegmentType(v2Name), SegmentName.RealtimeSegmentType.LLC);
-
- // Check table name
- assertEquals(longNameSegment.getTableName(), "myTable_REALTIME");
- assertEquals(shortNameSegment.getTableName(), "myTable_REALTIME");
- assertEquals(llcSegment.getTableName(), "myTable");
-
- // Check partition range
- assertEquals(longNameSegment.getPartitionRange(), "0");
- assertEquals(shortNameSegment.getPartitionRange(), "ALL");
- assertEquals(llcSegment.getPartitionGroupId(), 0);
-
- // Check groupId
- assertEquals(longNameSegment.getGroupId(), "myTable_REALTIME_1234567_0");
- assertEquals(shortNameSegment.getGroupId(), "myTable_REALTIME_1234567_0");
- try {
- llcSegment.getGroupId();
- fail("extractGroupIdName with a v2 name did not throw an exception");
- } catch (Exception e) {
- // Expected
- }
-
- // Check sequence number
- assertEquals(longNameSegment.getSequenceNumber(), 23456789);
- assertEquals(longNameSegment.getSequenceNumberStr(), "23456789");
- assertEquals(shortNameSegment.getSequenceNumberStr(), "1234567");
- assertEquals(shortNameSegment.getSequenceNumber(), 1234567);
- assertEquals(llcSegment.getSequenceNumber(), 1);
-
- assertEquals(llcSegment.getSegmentType(), SegmentName.RealtimeSegmentType.LLC);
- assertEquals(longNameSegment.getSegmentType(), SegmentName.RealtimeSegmentType.HLC_LONG);
- assertEquals(shortNameSegment.getSegmentType(), SegmentName.RealtimeSegmentType.HLC_SHORT);
-
- assertEquals(SegmentName.getSegmentType(llcSegment.getSegmentName()), SegmentName.RealtimeSegmentType.LLC);
- assertEquals(SegmentName.getSegmentType(longNameSegment.getSegmentName()),
- SegmentName.RealtimeSegmentType.HLC_LONG);
- assertEquals(SegmentName.getSegmentType(shortNameSegment.getSegmentName()),
- SegmentName.RealtimeSegmentType.HLC_SHORT);
-
- // Invalid segment names
- assertEquals(SegmentName.getSegmentType(longNameSegment.getSegmentName() + "__"),
- SegmentName.RealtimeSegmentType.UNSUPPORTED);
- assertEquals(SegmentName.getSegmentType("a__abc__1__3__4__54__g__gg___h"),
- SegmentName.RealtimeSegmentType.UNSUPPORTED);
- }
-
- @Test
- public void testExceptions() {
- // Nulls should not work
- /*
- try {
- SegmentNameBuilder.Realtime.buildHighLevelConsumerSegmentName(null, null, null);
- fail("Exception not thrown");
- } catch (Exception e) {
- // Expected
- }
- */
-
- // Double underscores should not work
- try {
- HLCSegmentName segName = new HLCSegmentName("__", "a", "b");
-// SegmentNameBuilder.Realtime.buildHighLevelConsumerSegmentName("__", "a", "b");
- fail("Exception not thrown");
- } catch (Exception e) {
- // Expected
- }
- }
-
- @Test
- public void testLLCSegName() {
- final String tableName = "myTable";
- final int partitionGroupId = 4;
- final int sequenceNumber = 27;
- final long msSinceEpoch = 1466200248000L;
- final String creationTime = "20160617T2150Z";
- final long creationTimeInMs = 1466200200000L;
- final String segmentName = "myTable__4__27__" + creationTime;
-
- LLCSegmentName segName1 = new LLCSegmentName(tableName, partitionGroupId, sequenceNumber, msSinceEpoch);
- Assert.assertEquals(segName1.getSegmentName(), segmentName);
- Assert.assertEquals(segName1.getPartitionGroupId(), partitionGroupId);
- Assert.assertEquals(segName1.getCreationTime(), creationTime);
- Assert.assertEquals(segName1.getCreationTimeMs(), creationTimeInMs);
- Assert.assertEquals(segName1.getSequenceNumber(), sequenceNumber);
- Assert.assertEquals(segName1.getTableName(), tableName);
-
- LLCSegmentName segName2 = new LLCSegmentName(segmentName);
- Assert.assertEquals(segName2.getSegmentName(), segmentName);
- Assert.assertEquals(segName2.getPartitionGroupId(), partitionGroupId);
- Assert.assertEquals(segName2.getCreationTime(), creationTime);
- Assert.assertEquals(segName2.getCreationTimeMs(), creationTimeInMs);
- Assert.assertEquals(segName2.getSequenceNumber(), sequenceNumber);
- Assert.assertEquals(segName2.getTableName(), tableName);
-
- Assert.assertEquals(segName1, segName2);
-
- LLCSegmentName segName3 = new LLCSegmentName(tableName, partitionGroupId + 1, sequenceNumber - 1, msSinceEpoch);
- Assert.assertTrue(segName1.compareTo(segName3) < 0);
- LLCSegmentName segName4 = new LLCSegmentName(tableName, partitionGroupId + 1, sequenceNumber + 1, msSinceEpoch);
- Assert.assertTrue(segName1.compareTo(segName4) < 0);
- LLCSegmentName segName5 = new LLCSegmentName(tableName, partitionGroupId - 1, sequenceNumber + 1, msSinceEpoch);
- Assert.assertTrue(segName1.compareTo(segName5) > 0);
- LLCSegmentName segName6 = new LLCSegmentName(tableName, partitionGroupId, sequenceNumber + 1, msSinceEpoch);
- Assert.assertTrue(segName1.compareTo(segName6) < 0);
-
- LLCSegmentName segName7 =
- new LLCSegmentName(tableName + "NotGood", partitionGroupId, sequenceNumber + 1, msSinceEpoch);
- try {
- segName1.compareTo(segName7);
- Assert.fail("Not failing when comparing " + segName1.getSegmentName() + " and " + segName7.getSegmentName());
- } catch (Exception e) {
- // expected
- }
- LLCSegmentName[] testSorted = new LLCSegmentName[]{segName3, segName1, segName4, segName5, segName6};
- Arrays.sort(testSorted);
- Assert.assertTrue(testSorted[0] == segName5);
- Assert.assertTrue(testSorted[1] == segName1);
- Assert.assertTrue(testSorted[2] == segName6);
- Assert.assertTrue(testSorted[3] == segName3);
- Assert.assertTrue(testSorted[4] == segName4);
- }
-}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigTest.java
index cef1b0cc0e..5d10893972 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigTest.java
@@ -22,8 +22,6 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.util.HashMap;
-import java.util.Map;
import java.util.stream.Stream;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -38,10 +36,7 @@ import static org.testng.AssertJUnit.assertTrue;
public class TableConfigTest {
-
- private static final String TEST_OFFLINE_TABLE_NAME = "testllc_OFFLINE";
- private static final String TEST_REALTIME_HLC_TABLE_NAME = "testhlc_REALTIME";
- private static final String TEST_REALTIME_LLC_TABLE_NAME = "testllc_REALTIME";
+ private static final String RAW_TABLE_NAME = "testTable";
@DataProvider
public Object[][] configs()
@@ -67,34 +62,19 @@ public class TableConfigTest {
@Test
public void testGetReplication() {
TableConfig offlineTableConfig =
- new TableConfigBuilder(TableType.OFFLINE).setTableName(TEST_OFFLINE_TABLE_NAME).setNumReplicas(2).build();
- assertEquals(2, offlineTableConfig.getReplication());
-
+ new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).setNumReplicas(2).build();
+ assertEquals(offlineTableConfig.getReplication(), 2);
offlineTableConfig.getValidationConfig().setReplication("4");
- assertEquals(4, offlineTableConfig.getReplication());
-
+ assertEquals(offlineTableConfig.getReplication(), 4);
offlineTableConfig.getValidationConfig().setReplicasPerPartition("3");
- assertEquals(4, offlineTableConfig.getReplication());
-
- TableConfig realtimeLLCTableConfig =
- new TableConfigBuilder(TableType.REALTIME).setTableName(TEST_REALTIME_LLC_TABLE_NAME)
- .setStreamConfigs(getStreamConfigMap("lowlevel")).setLLC(true).setNumReplicas(2).build();
-
- assertEquals(2, realtimeLLCTableConfig.getReplication());
-
- realtimeLLCTableConfig.getValidationConfig().setReplication("4");
- assertEquals(2, realtimeLLCTableConfig.getReplication());
-
- realtimeLLCTableConfig.getValidationConfig().setReplicasPerPartition("3");
- assertEquals(3, realtimeLLCTableConfig.getReplication());
- }
+ assertEquals(offlineTableConfig.getReplication(), 4);
- private Map<String, String> getStreamConfigMap(String consumerType) {
- Map<String, String> configMap = new HashMap<>();
- configMap.put("streamType", "kafka");
- configMap.put("stream.kafka.consumer.type", consumerType);
- configMap.put("stream.kafka.topic.name", "test");
- configMap.put("stream.kafka.decoder.class.name", "test");
- return configMap;
+ TableConfig realtimeTableConfig =
+ new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(2).build();
+ assertEquals(realtimeTableConfig.getReplication(), 2);
+ realtimeTableConfig.getValidationConfig().setReplication("4");
+ assertEquals(realtimeTableConfig.getReplication(), 4);
+ realtimeTableConfig.getValidationConfig().setReplicasPerPartition("3");
+ assertEquals(realtimeTableConfig.getReplication(), 3);
}
}
diff --git a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigUtilsTest.java b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigUtilsTest.java
index edb3b58f98..2ff578b5ee 100644
--- a/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigUtilsTest.java
+++ b/pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigUtilsTest.java
@@ -31,7 +31,6 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.ingestion.BatchIngestionConfig;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamMessageDecoder;
@@ -171,25 +170,19 @@ public class TableConfigUtilsTest {
private Map<String, String> getTestStreamConfigs() {
String streamType = "testStream";
String topic = "testTopic";
- String consumerType = StreamConfig.ConsumerType.LOWLEVEL.toString();
String consumerFactoryClass = TestStreamConsumerFactory.class.getName();
String decoderClass = TestStreamMessageDecoder.class.getName();
// All mandatory properties set
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, "streamType");
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
return streamConfigMap;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java
index 10a9114149..0bf2da9314 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotSegmentRestletResource.java
@@ -69,7 +69,7 @@ import org.apache.pinot.common.lineage.SegmentLineage;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.controllerjob.ControllerJobType;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.utils.SegmentName;
+import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.api.access.AccessType;
@@ -510,12 +510,13 @@ public class PinotSegmentRestletResource {
/**
* Helper method to find the existing table based on the given table name (with or without type suffix) and segment
* name.
+ * TODO: Real-time table might also contain uploaded segments (not with LLC name), which is not handled here.
*/
private String getExistingTable(String tableName, String segmentName) {
TableType tableType = TableNameBuilder.getTableTypeFromTableName(tableName);
if (tableType == null) {
// Derive table type from segment name if the given table name doesn't have type suffix
- tableType = SegmentName.isRealtimeSegmentName(segmentName) ? TableType.REALTIME : TableType.OFFLINE;
+ tableType = LLCSegmentName.isLLCSegment(segmentName) ? TableType.REALTIME : TableType.OFFLINE;
}
return ResourceUtils.getExistingTableNamesWithType(_pinotHelixResourceManager, tableName, tableType, LOGGER).get(0);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
index 619a82cdbe..f64e6c3e75 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/SegmentStatusChecker.java
@@ -47,7 +47,7 @@ import org.apache.pinot.controller.helix.core.realtime.MissingConsumingSegmentFi
import org.apache.pinot.controller.util.TableSizeReader;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
@@ -311,10 +311,10 @@ public class SegmentStatusChecker extends ControllerPeriodicTask<SegmentStatusCh
}
if (tableType == TableType.REALTIME && tableConfig != null) {
- PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig));
- new MissingConsumingSegmentFinder(tableNameWithType, propertyStore, _controllerMetrics, streamConfig)
- .findAndEmitMetrics(idealState);
+ StreamConfig streamConfig =
+ new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ new MissingConsumingSegmentFinder(tableNameWithType, propertyStore, _controllerMetrics,
+ streamConfig).findAndEmitMetrics(idealState);
}
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
index 1974d952b5..ae5b271238 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java
@@ -120,7 +120,6 @@ import org.apache.pinot.common.tier.TierSegmentSelector;
import org.apache.pinot.common.utils.BcryptUtils;
import org.apache.pinot.common.utils.HashUtil;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.config.AccessControlUserConfigUtils;
import org.apache.pinot.common.utils.config.InstanceUtils;
import org.apache.pinot.common.utils.config.TableConfigUtils;
@@ -150,7 +149,6 @@ import org.apache.pinot.controller.helix.core.rebalance.ZkBasedTableRebalanceObs
import org.apache.pinot.controller.helix.core.util.ZKMetadataUtils;
import org.apache.pinot.controller.helix.starter.HelixConfig;
import org.apache.pinot.segment.spi.SegmentMetadata;
-import org.apache.pinot.spi.config.ConfigUtils;
import org.apache.pinot.spi.config.instance.Instance;
import org.apache.pinot.spi.config.table.IndexingConfig;
import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
@@ -169,7 +167,6 @@ import org.apache.pinot.spi.config.user.RoleType;
import org.apache.pinot.spi.config.user.UserConfig;
import org.apache.pinot.spi.data.DateTimeFieldSpec;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
import org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.BrokerResourceStateModel;
@@ -894,16 +891,15 @@ public class PinotHelixResourceManager {
public Collection<String> getLastLLCCompletedSegments(String tableNameWithType) {
Map<Integer, String> partitionIdToLastLLCCompletedSegmentMap = new HashMap<>();
- for (SegmentZKMetadata segMetadata : getSegmentsZKMetadata(tableNameWithType)) {
- if (SegmentName.isLowLevelConsumerSegmentName(segMetadata.getSegmentName())
- && segMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE) {
- LLCSegmentName llcName = LLCSegmentName.of(segMetadata.getSegmentName());
+ for (SegmentZKMetadata zkMetadata : getSegmentsZKMetadata(tableNameWithType)) {
+ if (zkMetadata.getStatus() == CommonConstants.Segment.Realtime.Status.DONE) {
+ LLCSegmentName llcName = LLCSegmentName.of(zkMetadata.getSegmentName());
int partitionGroupId = llcName.getPartitionGroupId();
int sequenceNumber = llcName.getSequenceNumber();
String lastCompletedSegName = partitionIdToLastLLCCompletedSegmentMap.get(partitionGroupId);
if (lastCompletedSegName == null
|| LLCSegmentName.of(lastCompletedSegName).getSequenceNumber() < sequenceNumber) {
- partitionIdToLastLLCCompletedSegmentMap.put(partitionGroupId, segMetadata.getSegmentName());
+ partitionIdToLastLLCCompletedSegmentMap.put(partitionGroupId, zkMetadata.getSegmentName());
}
}
}
@@ -1551,97 +1547,64 @@ public class PinotHelixResourceManager {
+ "If the external view is not removed after a long time, try restarting the servers showing up in the "
+ "external view");
}
-
validateTableTenantConfig(tableConfig);
- TableType tableType = tableConfig.getTableType();
-
- switch (tableType) {
- case OFFLINE:
- // now lets build an ideal state
- LOGGER.info("building empty ideal state for table : " + tableNameWithType);
- IdealState offlineIdealState =
- PinotTableIdealStateBuilder.buildEmptyIdealStateFor(tableNameWithType, tableConfig.getReplication(),
- _enableBatchMessageMode);
- LOGGER.info("adding table via the admin");
-
- try {
- _helixAdmin.addResource(_helixClusterName, tableNameWithType, offlineIdealState);
-
- // lets add table configs
- ZKMetadataProvider.setOfflineTableConfig(_propertyStore, tableNameWithType,
- TableConfigUtils.toZNRecord(tableConfig));
-
- // Assign instances
- assignInstances(tableConfig, true);
- } catch (Exception e) {
- LOGGER.error("Caught exception during offline table setup. Cleaning up table {}", tableNameWithType, e);
- deleteOfflineTable(tableNameWithType);
- throw e;
- }
-
- LOGGER.info("Successfully added table: {}", tableNameWithType);
- break;
-
- case REALTIME:
- verifyStreamConfig(tableNameWithType, tableConfig);
-
- // Ensure that realtime table is not created if schema is not present
- Schema schema =
- ZKMetadataProvider.getSchema(_propertyStore, TableNameBuilder.extractRawTableName(tableNameWithType));
-
- if (schema == null) {
- // Fall back to getting schema-name from table config if schema-name != table-name
- String schemaName = tableConfig.getValidationConfig().getSchemaName();
- if (schemaName == null || ZKMetadataProvider.getSchema(_propertyStore, schemaName) == null) {
- throw new InvalidTableConfigException("No schema defined for realtime table: " + tableNameWithType);
- }
- }
- try {
- // lets add table configs
- ZKMetadataProvider.setRealtimeTableConfig(_propertyStore, tableNameWithType,
- TableConfigUtils.toZNRecord(tableConfig));
-
- // Assign instances before setting up the real-time cluster so that new LLC CONSUMING segment can be assigned
- // based on the instance partitions
- assignInstances(tableConfig, true);
-
- /*
- * PinotRealtimeSegmentManager sets up watches on table and segment path. When a table gets created,
- * it expects the INSTANCE path in propertystore to be set up so that it can get the group ID and
- * create (high-level consumer) segments for that table.
- * So, we need to set up the instance first, before adding the table resource for HLC new table creation.
- *
- * For low-level consumers, the order is to create the resource first, and set up the propertystore with
- * segments
- * and then tweak the idealstate to add those segments.
- *
- * We also need to support the case when a high-level consumer already exists for a table and we are adding
- * the low-level consumers.
- */
- ensureRealtimeClusterIsSetUp(tableConfig);
- } catch (Exception e) {
- LOGGER.error("Caught exception during realtime table setup. Cleaning up table {}", tableNameWithType, e);
- deleteRealtimeTable(tableNameWithType);
- throw e;
+ IdealState idealState =
+ PinotTableIdealStateBuilder.buildEmptyIdealStateFor(tableNameWithType, tableConfig.getReplication(),
+ _enableBatchMessageMode);
+ TableType tableType = tableConfig.getTableType();
+ if (tableType == TableType.OFFLINE) {
+ try {
+ // Add table config
+ ZKMetadataProvider.setTableConfig(_propertyStore, tableNameWithType, TableConfigUtils.toZNRecord(tableConfig));
+ // Assign instances
+ assignInstances(tableConfig, true);
+ // Add ideal state
+ _helixAdmin.addResource(_helixClusterName, tableNameWithType, idealState);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception during offline table setup. Cleaning up table {}", tableNameWithType, e);
+ deleteOfflineTable(tableNameWithType);
+ throw e;
+ }
+ } else {
+ Preconditions.checkState(tableType == TableType.REALTIME, "Invalid table type: %s", tableType);
+
+ // Ensure that realtime table is not created if schema is not present
+ Schema schema =
+ ZKMetadataProvider.getSchema(_propertyStore, TableNameBuilder.extractRawTableName(tableNameWithType));
+ if (schema == null) {
+ // Fall back to getting schema-name from table config if schema-name != table-name
+ String schemaName = tableConfig.getValidationConfig().getSchemaName();
+ if (schemaName == null || ZKMetadataProvider.getSchema(_propertyStore, schemaName) == null) {
+ throw new InvalidTableConfigException("No schema defined for realtime table: " + tableNameWithType);
}
+ }
- LOGGER.info("Successfully added or updated the table {} ", tableNameWithType);
- break;
-
- default:
- throw new InvalidTableConfigException("Unsupported table type: " + tableType);
+ try {
+ // Add table config
+ ZKMetadataProvider.setTableConfig(_propertyStore, tableNameWithType, TableConfigUtils.toZNRecord(tableConfig));
+ // Assign instances
+ assignInstances(tableConfig, true);
+ // Add ideal state with the first CONSUMING segment
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(tableConfig, idealState);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception during realtime table setup. Cleaning up table {}", tableNameWithType, e);
+ deleteRealtimeTable(tableNameWithType);
+ throw e;
+ }
}
- LOGGER.info("Updating BrokerResource IdealState for table: {}", tableNameWithType);
+ LOGGER.info("Updating BrokerResource for table: {}", tableNameWithType);
List<String> brokers =
HelixHelper.getInstancesWithTag(_helixZkManager, TagNameUtils.extractBrokerTag(tableConfig.getTenantConfig()));
- HelixHelper.updateIdealState(_helixZkManager, Helix.BROKER_RESOURCE_INSTANCE, idealState -> {
- assert idealState != null;
- idealState.getRecord().getMapFields()
+ HelixHelper.updateIdealState(_helixZkManager, Helix.BROKER_RESOURCE_INSTANCE, is -> {
+ assert is != null;
+ is.getRecord().getMapFields()
.put(tableNameWithType, SegmentAssignmentUtils.getInstanceStateMap(brokers, BrokerResourceStateModel.ONLINE));
- return idealState;
+ return is;
});
+
+ LOGGER.info("Successfully added table: {}", tableNameWithType);
}
/**
@@ -1757,62 +1720,6 @@ public class PinotHelixResourceManager {
_pinotLLCRealtimeSegmentManager = pinotLLCRealtimeSegmentManager;
}
- private void verifyStreamConfig(String tableNameWithType, TableConfig tableConfig) {
- // Check if HLC table is allowed.
- StreamConfig streamConfig =
- new StreamConfig(tableNameWithType, IngestionConfigUtils.getStreamConfigMap(tableConfig));
- if (streamConfig.hasHighLevelConsumerType()) {
- throw new InvalidTableConfigException(
- "Creating HLC realtime table is not allowed for Table: " + tableNameWithType);
- }
- }
-
- private void ensureRealtimeClusterIsSetUp(TableConfig rawRealtimeTableConfig) {
- // Need to apply environment variabls here to ensure the secrets used in stream configs are correctly applied.
- TableConfig realtimeTableConfig = ConfigUtils.applyConfigWithEnvVariables(rawRealtimeTableConfig);
- String realtimeTableName = realtimeTableConfig.getTableName();
- StreamConfig streamConfig = new StreamConfig(realtimeTableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
- IdealState idealState = getTableIdealState(realtimeTableName);
-
- if (streamConfig.hasHighLevelConsumerType()) {
- if (idealState == null) {
- LOGGER.info("Initializing IdealState for HLC table: {}", realtimeTableName);
- idealState = PinotTableIdealStateBuilder.buildInitialHighLevelRealtimeIdealStateFor(realtimeTableName,
- realtimeTableConfig, _helixZkManager, _propertyStore, _enableBatchMessageMode);
- _helixAdmin.addResource(_helixClusterName, realtimeTableName, idealState);
- } else {
- // Remove LLC segments if it is not configured
- if (!streamConfig.hasLowLevelConsumerType()) {
- _pinotLLCRealtimeSegmentManager.removeLLCSegments(idealState);
- }
- }
- // For HLC table, property store entry must exist to trigger watchers to create segments
- ensurePropertyStoreEntryExistsForHighLevelConsumer(realtimeTableName);
- }
-
- // Either we have only low-level consumer, or both.
- if (streamConfig.hasLowLevelConsumerType()) {
- // Will either create idealstate entry, or update the IS entry with new segments
- // (unless there are low-level segments already present)
- if (ZKMetadataProvider.getLLCRealtimeSegments(_propertyStore, realtimeTableName).isEmpty()) {
- PinotTableIdealStateBuilder.buildLowLevelRealtimeIdealStateFor(_pinotLLCRealtimeSegmentManager,
- realtimeTableName, realtimeTableConfig, idealState, _enableBatchMessageMode);
- LOGGER.info("Successfully added Helix entries for low-level consumers for {} ", realtimeTableName);
- } else {
- LOGGER.info("LLC is already set up for table {}, not configuring again", realtimeTableName);
- }
- }
- }
-
- private void ensurePropertyStoreEntryExistsForHighLevelConsumer(String realtimeTableName) {
- String propertyStorePath = ZKMetadataProvider.constructPropertyStorePathForResource(realtimeTableName);
- if (!_propertyStore.exists(propertyStorePath, AccessOption.PERSISTENT)) {
- LOGGER.info("Creating property store entry for HLC table: {}", realtimeTableName);
- _propertyStore.create(propertyStorePath, new ZNRecord(realtimeTableName), AccessOption.PERSISTENT);
- }
- }
-
private void assignInstances(TableConfig tableConfig, boolean override) {
String tableNameWithType = tableConfig.getTableName();
String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
@@ -1895,46 +1802,22 @@ public class PinotHelixResourceManager {
public void setExistingTableConfig(TableConfig tableConfig)
throws IOException {
String tableNameWithType = tableConfig.getTableName();
- SegmentsValidationAndRetentionConfig segmentsConfig = tableConfig.getValidationConfig();
-
- TableType tableType = tableConfig.getTableType();
- switch (tableType) {
- case OFFLINE:
- ZKMetadataProvider.setOfflineTableConfig(_propertyStore, tableNameWithType,
- TableConfigUtils.toZNRecord(tableConfig));
-
- // Update IdealState replication
- IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableNameWithType);
- String replicationConfigured = Integer.toString(tableConfig.getReplication());
- if (!idealState.getReplicas().equals(replicationConfigured)) {
- HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, is -> {
- assert is != null;
- is.setReplicas(replicationConfigured);
- return is;
- }, RetryPolicies.exponentialBackoffRetryPolicy(5, 1000L, 1.2f));
- }
-
- // Assign instances
- assignInstances(tableConfig, false);
-
- break;
-
- case REALTIME:
- verifyStreamConfig(tableNameWithType, tableConfig);
- ZKMetadataProvider.setRealtimeTableConfig(_propertyStore, tableNameWithType,
- TableConfigUtils.toZNRecord(tableConfig));
-
- // Assign instances before setting up the real-time cluster so that new LLC CONSUMING segment can be assigned
- // based on the instance partitions
- assignInstances(tableConfig, false);
- ensureRealtimeClusterIsSetUp(tableConfig);
-
- break;
+ ZKMetadataProvider.setTableConfig(_propertyStore, tableNameWithType, TableConfigUtils.toZNRecord(tableConfig));
- default:
- throw new InvalidTableConfigException("Unsupported table type: " + tableType);
+ // Update IdealState replication
+ IdealState idealState = _helixAdmin.getResourceIdealState(_helixClusterName, tableNameWithType);
+ String replicationConfigured = Integer.toString(tableConfig.getReplication());
+ if (!idealState.getReplicas().equals(replicationConfigured)) {
+ HelixHelper.updateIdealState(_helixZkManager, tableNameWithType, is -> {
+ assert is != null;
+ is.setReplicas(replicationConfigured);
+ return is;
+ }, RetryPolicies.exponentialBackoffRetryPolicy(5, 1000L, 1.2f));
}
+ // Assign instances
+ assignInstances(tableConfig, false);
+
// Send update query quota message if quota is specified
sendTableConfigRefreshMessage(tableNameWithType);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
index ac24151d67..23a115417f 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotTableIdealStateBuilder.java
@@ -19,105 +19,38 @@
package org.apache.pinot.controller.helix.core;
import java.util.List;
-import java.util.Map;
-import org.apache.helix.HelixManager;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.builder.CustomModeISBuilder;
-import org.apache.helix.store.zk.ZkHelixPropertyStore;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
-import org.apache.pinot.common.utils.config.TagNameUtils;
-import org.apache.pinot.common.utils.helix.HelixHelper;
-import org.apache.pinot.controller.api.exception.InvalidTableConfigException;
-import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
-import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionGroupMetadataFetcher;
import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
-import org.apache.pinot.spi.utils.StringUtil;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
import org.apache.pinot.spi.utils.retry.RetryPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * Pinot data server layer IdealState builder.
- *
- *
- */
public class PinotTableIdealStateBuilder {
private PinotTableIdealStateBuilder() {
}
private static final Logger LOGGER = LoggerFactory.getLogger(PinotTableIdealStateBuilder.class);
- public static final String ONLINE = "ONLINE";
- public static final String OFFLINE = "OFFLINE";
private static final RetryPolicy DEFAULT_IDEALSTATE_UPDATE_RETRY_POLICY =
RetryPolicies.randomDelayRetryPolicy(3, 100L, 200L);
- /**
- *
- * Building an empty idealState for a given table.
- * Used when creating a new table.
- *
- * @param tableName resource name
- * @param numCopies is the number of replicas
- * @return
- */
- public static IdealState buildEmptyIdealStateFor(String tableName, int numCopies, boolean enableBatchMessageMode) {
- final CustomModeISBuilder customModeIdealStateBuilder = new CustomModeISBuilder(tableName);
- final int replicas = numCopies;
+ public static IdealState buildEmptyIdealStateFor(String tableNameWithType, int numReplicas,
+ boolean enableBatchMessageMode) {
+ CustomModeISBuilder customModeIdealStateBuilder = new CustomModeISBuilder(tableNameWithType);
customModeIdealStateBuilder
.setStateModel(PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL)
- .setNumPartitions(0).setNumReplica(replicas).setMaxPartitionsPerNode(1);
- final IdealState idealState = customModeIdealStateBuilder.build();
- idealState.setInstanceGroupTag(tableName);
+ .setNumPartitions(0).setNumReplica(numReplicas).setMaxPartitionsPerNode(1);
+ IdealState idealState = customModeIdealStateBuilder.build();
+ idealState.setInstanceGroupTag(tableNameWithType);
idealState.setBatchMessageMode(enableBatchMessageMode);
return idealState;
}
- public static IdealState addNewRealtimeSegmentToIdealState(String segmentId, IdealState state, String instanceName) {
- state.setPartitionState(segmentId, instanceName, ONLINE);
- state.setNumPartitions(state.getNumPartitions() + 1);
- return state;
- }
-
- public static IdealState buildInitialHighLevelRealtimeIdealStateFor(String realtimeTableName,
- TableConfig realtimeTableConfig, HelixManager helixManager, ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore,
- boolean enableBatchMessageMode) {
- List<String> realtimeInstances = HelixHelper.getInstancesWithTag(helixManager,
- TagNameUtils.extractConsumingServerTag(realtimeTableConfig.getTenantConfig()));
- IdealState idealState = buildEmptyRealtimeIdealStateFor(realtimeTableName, 1, enableBatchMessageMode);
- if (realtimeInstances.size() % realtimeTableConfig.getReplication() != 0) {
- throw new RuntimeException(
- "Number of instance in current tenant should be an integer multiples of the number of replications");
- }
- setupInstanceConfigForHighLevelConsumer(realtimeTableName, realtimeInstances.size(),
- realtimeTableConfig.getReplication(), IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig),
- zkHelixPropertyStore, realtimeInstances);
- return idealState;
- }
-
- public static void buildLowLevelRealtimeIdealStateFor(PinotLLCRealtimeSegmentManager pinotLLCRealtimeSegmentManager,
- String realtimeTableName, TableConfig realtimeTableConfig, IdealState idealState,
- boolean enableBatchMessageMode) {
- // Validate replicasPerPartition here.
- final int nReplicas;
- try {
- nReplicas = realtimeTableConfig.getReplication();
- } catch (NumberFormatException e) {
- throw new InvalidTableConfigException("Invalid value for replicasPerPartition, expected a number.", e);
- }
- if (idealState == null) {
- idealState = buildEmptyRealtimeIdealStateFor(realtimeTableName, nReplicas, enableBatchMessageMode);
- }
- pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
- }
-
/**
* Fetches the list of {@link PartitionGroupMetadata} for the new partition groups for the stream,
* with the help of the {@link PartitionGroupConsumptionStatus} of the current partitionGroups.
@@ -166,56 +99,4 @@ public class PinotTableIdealStateBuilder {
throw new RuntimeException(fetcherException);
}
}
-
- public static IdealState buildEmptyRealtimeIdealStateFor(String realtimeTableName, int replicaCount,
- boolean enableBatchMessageMode) {
- final CustomModeISBuilder customModeIdealStateBuilder = new CustomModeISBuilder(realtimeTableName);
- customModeIdealStateBuilder
- .setStateModel(PinotHelixSegmentOnlineOfflineStateModelGenerator.PINOT_SEGMENT_ONLINE_OFFLINE_STATE_MODEL)
- .setNumPartitions(0).setNumReplica(replicaCount).setMaxPartitionsPerNode(1);
- final IdealState idealState = customModeIdealStateBuilder.build();
- idealState.setInstanceGroupTag(realtimeTableName);
- idealState.setBatchMessageMode(enableBatchMessageMode);
-
- return idealState;
- }
-
- private static void setupInstanceConfigForHighLevelConsumer(String realtimeTableName, int numDataInstances,
- int numDataReplicas, Map<String, String> streamConfig, ZkHelixPropertyStore<ZNRecord> zkHelixPropertyStore,
- List<String> instanceList) {
- int numInstancesPerReplica = numDataInstances / numDataReplicas;
- int partitionId = 0;
- int replicaId = 0;
-
- String groupId = getGroupIdFromRealtimeDataTable(realtimeTableName, streamConfig);
- for (int i = 0; i < numInstancesPerReplica * numDataReplicas; i++) {
- String instance = instanceList.get(i);
- InstanceZKMetadata instanceZKMetadata = ZKMetadataProvider.getInstanceZKMetadata(zkHelixPropertyStore, instance);
- if (instanceZKMetadata == null) {
- instanceZKMetadata = new InstanceZKMetadata();
- String[] instanceConfigs = instance.split("_");
- assert (instanceConfigs.length == 3);
- instanceZKMetadata.setInstanceType(instanceConfigs[0]);
- instanceZKMetadata.setInstanceName(instanceConfigs[1]);
- instanceZKMetadata.setInstancePort(Integer.parseInt(instanceConfigs[2]));
- }
- instanceZKMetadata.setGroupId(realtimeTableName, groupId + "_" + replicaId);
- instanceZKMetadata.setPartition(realtimeTableName, Integer.toString(partitionId));
- partitionId = (partitionId + 1) % numInstancesPerReplica;
- if (partitionId == 0) {
- replicaId++;
- }
- ZKMetadataProvider.setInstanceZKMetadata(zkHelixPropertyStore, instanceZKMetadata);
- }
- }
-
- private static String getGroupIdFromRealtimeDataTable(String realtimeTableName, Map<String, String> streamConfigMap) {
- String groupId = StringUtil.join("_", realtimeTableName, System.currentTimeMillis() + "");
- StreamConfig streamConfig = new StreamConfig(realtimeTableName, streamConfigMap);
- String streamConfigGroupId = streamConfig.getGroupId();
- if (streamConfigGroupId != null && !streamConfigGroupId.isEmpty()) {
- groupId = streamConfigGroupId;
- }
- return groupId;
- }
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
index ebde702b09..49a47ec9bd 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java
@@ -42,7 +42,6 @@ import org.apache.helix.model.IdealState;
import org.apache.helix.store.zk.ZkHelixPropertyStore;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.core.segment.processing.lifecycle.PinotSegmentLifecycleEventListenerManager;
@@ -216,10 +215,6 @@ public class SegmentDeletionManager {
protected void removeSegmentFromStore(String tableNameWithType, String segmentId,
@Nullable Long deletedSegmentsRetentionMs) {
- // Ignore HLC segments as they are not stored in Pinot FS
- if (SegmentName.isHighLevelConsumerSegmentName(segmentId)) {
- return;
- }
if (_dataDir != null) {
long retentionMs = deletedSegmentsRetentionMs == null
? _defaultDeletedSegmentsRetentionMs : deletedSegmentsRetentionMs;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
index d95e058239..f4192a5a1a 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/MissingConsumingSegmentFinder.java
@@ -37,7 +37,7 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
import org.apache.pinot.spi.stream.OffsetCriteria;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
@@ -65,7 +65,7 @@ public class MissingConsumingSegmentFinder {
private ControllerMetrics _controllerMetrics;
public MissingConsumingSegmentFinder(String realtimeTableName, ZkHelixPropertyStore<ZNRecord> propertyStore,
- ControllerMetrics controllerMetrics, PartitionLevelStreamConfig streamConfig) {
+ ControllerMetrics controllerMetrics, StreamConfig streamConfig) {
_realtimeTableName = realtimeTableName;
_controllerMetrics = controllerMetrics;
_segmentMetadataFetcher = new SegmentMetadataFetcher(propertyStore, controllerMetrics);
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 11d82875cf..e55f0e6463 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
@@ -56,7 +56,6 @@ import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.protocols.SegmentCompletionProtocol;
import org.apache.pinot.common.utils.FileUploadDownloadClient;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.controller.ControllerConf;
@@ -91,7 +90,6 @@ import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.OffsetCriteria;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
@@ -302,18 +300,10 @@ public class PinotLLCRealtimeSegmentManager {
String realtimeTableName = tableConfig.getTableName();
LOGGER.info("Setting up new LLC table: {}", realtimeTableName);
- // Make sure all the existing segments are HLC segments
- List<String> currentSegments = getAllSegments(realtimeTableName);
- for (String segmentName : currentSegments) {
- // TODO: Should return 4xx HTTP status code. Currently all exceptions are returning 500
- Preconditions.checkState(SegmentName.isHighLevelConsumerSegmentName(segmentName),
- "Cannot set up new LLC table: %s with existing non-HLC segment: %s", realtimeTableName, segmentName);
- }
-
_flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName);
- PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ StreamConfig streamConfig =
+ new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
List<PartitionGroupMetadata> newPartitionGroupMetadataList =
getNewPartitionGroupMetadataList(streamConfig, Collections.emptyList());
@@ -338,24 +328,6 @@ public class PinotLLCRealtimeSegmentManager {
setIdealState(realtimeTableName, idealState);
}
- /**
- * Removes all LLC segments from the given IdealState.
- */
- public void removeLLCSegments(IdealState idealState) {
- Preconditions.checkState(!_isStopping, "Segment manager is stopping");
-
- String realtimeTableName = idealState.getResourceName();
- LOGGER.info("Removing LLC segments for table: {}", realtimeTableName);
-
- List<String> segmentsToRemove = new ArrayList<>();
- for (String segmentName : idealState.getRecord().getMapFields().keySet()) {
- if (SegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- segmentsToRemove.add(segmentName);
- }
- }
- _helixResourceManager.deleteSegments(realtimeTableName, segmentsToRemove);
- }
-
// TODO: Consider using TableCache to read the table config
@VisibleForTesting
public TableConfig getTableConfig(String realtimeTableName) {
@@ -551,8 +523,8 @@ public class PinotLLCRealtimeSegmentManager {
_helixResourceManager.sendSegmentRefreshMessage(realtimeTableName, committingSegmentName, false, true);
// Using the latest segment of each partition group, creates a list of {@link PartitionGroupConsumptionStatus}
- PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ StreamConfig streamConfig =
+ new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
List<PartitionGroupConsumptionStatus> currentPartitionGroupConsumptionStatusList =
getPartitionGroupConsumptionStatusList(idealState, streamConfig);
@@ -673,7 +645,7 @@ public class PinotLLCRealtimeSegmentManager {
/**
* Creates and persists segment ZK metadata for the new CONSUMING segment.
*/
- private void createNewSegmentZKMetadata(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig,
+ private void createNewSegmentZKMetadata(TableConfig tableConfig, StreamConfig streamConfig,
LLCSegmentName newLLCSegmentName, long creationTimeMs, CommittingSegmentDescriptor committingSegmentDescriptor,
@Nullable SegmentZKMetadata committingSegmentZKMetadata, InstancePartitions instancePartitions,
int numPartitionGroups, int numReplicas, List<PartitionGroupMetadata> partitionGroupMetadataList) {
@@ -885,7 +857,7 @@ public class PinotLLCRealtimeSegmentManager {
* (this operation is done only if @param recreateDeletedConsumingSegment is set to true,
* which means it's manually triggered by admin not by automatic periodic task)
*/
- public void ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig,
+ public void ensureAllPartitionsConsuming(TableConfig tableConfig, StreamConfig streamConfig,
boolean recreateDeletedConsumingSegment, OffsetCriteria offsetCriteria) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
@@ -1066,9 +1038,9 @@ public class PinotLLCRealtimeSegmentManager {
* TODO: split this method into multiple smaller methods
*/
@VisibleForTesting
- IdealState ensureAllPartitionsConsuming(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig,
- IdealState idealState, List<PartitionGroupMetadata> newPartitionGroupMetadataList,
- boolean recreateDeletedConsumingSegment, OffsetCriteria offsetCriteria) {
+ IdealState ensureAllPartitionsConsuming(TableConfig tableConfig, StreamConfig streamConfig, IdealState idealState,
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList, boolean recreateDeletedConsumingSegment,
+ OffsetCriteria offsetCriteria) {
String realtimeTableName = tableConfig.getTableName();
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
@@ -1261,7 +1233,7 @@ public class PinotLLCRealtimeSegmentManager {
return idealState;
}
- private void createNewConsumingSegment(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig,
+ private void createNewConsumingSegment(TableConfig tableConfig, StreamConfig streamConfig,
SegmentZKMetadata latestSegmentZKMetadata, long currentTimeMs,
List<PartitionGroupMetadata> newPartitionGroupMetadataList, InstancePartitions instancePartitions,
Map<String, Map<String, String>> instanceStatesMap, SegmentAssignment segmentAssignment,
@@ -1323,7 +1295,7 @@ public class PinotLLCRealtimeSegmentManager {
* Sets up a new partition group.
* <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
*/
- private String setupNewPartitionGroup(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig,
+ private String setupNewPartitionGroup(TableConfig tableConfig, StreamConfig streamConfig,
PartitionGroupMetadata partitionGroupMetadata, long creationTimeMs, InstancePartitions instancePartitions,
int numPartitionGroups, int numReplicas, List<PartitionGroupMetadata> partitionGroupMetadataList) {
String realtimeTableName = tableConfig.getTableName();
@@ -1425,7 +1397,7 @@ public class PinotLLCRealtimeSegmentManager {
String segmentName = segmentZKMetadata.getSegmentName();
try {
// Only fix the committed (DONE) LLC segment without deep store copy (empty download URL)
- if (!SegmentName.isLowLevelConsumerSegmentName(segmentName) || segmentZKMetadata.getStatus() != Status.DONE
+ if (segmentZKMetadata.getStatus() != Status.DONE
|| !CommonConstants.Segment.METADATA_URI_FOR_PEER_DOWNLOAD.equals(segmentZKMetadata.getDownloadUrl())) {
continue;
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
deleted file mode 100644
index 6e7e9281a6..0000000000
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotRealtimeSegmentManager.java
+++ /dev/null
@@ -1,421 +0,0 @@
-/**
- * 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.controller.helix.core.realtime;
-
-import com.google.common.base.Function;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.helix.PropertyPathBuilder;
-import org.apache.helix.model.IdealState;
-import org.apache.helix.store.HelixPropertyListener;
-import org.apache.helix.zookeeper.datamodel.ZNRecord;
-import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
-import org.apache.helix.zookeeper.impl.client.ZkClient;
-import org.apache.helix.zookeeper.zkclient.IZkChildListener;
-import org.apache.helix.zookeeper.zkclient.IZkDataListener;
-import org.apache.pinot.common.Utils;
-import org.apache.pinot.common.metadata.ZKMetadataProvider;
-import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
-import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.metrics.ControllerMeter;
-import org.apache.pinot.common.metrics.ControllerMetrics;
-import org.apache.pinot.common.utils.HLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
-import org.apache.pinot.common.utils.config.TableConfigUtils;
-import org.apache.pinot.common.utils.helix.HelixHelper;
-import org.apache.pinot.controller.ControllerConf;
-import org.apache.pinot.controller.LeadControllerManager;
-import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
-import org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder;
-import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.CommonConstants;
-import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
-import org.apache.pinot.spi.utils.builder.TableNameBuilder;
-import org.apache.pinot.spi.utils.retry.RetryPolicies;
-import org.apache.zookeeper.data.Stat;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Realtime segment manager, which assigns realtime segments to server instances so that they can consume from the
- * stream.
- */
-public class PinotRealtimeSegmentManager implements HelixPropertyListener, IZkChildListener, IZkDataListener {
- private static final Logger LOGGER = LoggerFactory.getLogger(PinotRealtimeSegmentManager.class);
- private static final String TABLE_CONFIG = "/CONFIGS/TABLE";
- private static final String SEGMENTS_PATH = "/SEGMENTS";
- private static final String REALTIME_SEGMENT_PROPERTY_STORE_PATH_PATTERN =
- ".*/SEGMENTS/.*_REALTIME|.*/SEGMENTS/.*_REALTIME/.*";
- private static final String REALTIME_TABLE_CONFIG_PROPERTY_STORE_PATH_PATTERN = ".*/TABLE/.*REALTIME";
- private static final String CONTROLLER_LEADER_CHANGE = "CONTROLLER LEADER CHANGE";
-
- private final String _propertyStorePath;
- private final String _tableConfigPath;
- private final PinotHelixResourceManager _pinotHelixResourceManager;
- private ZkClient _zkClient;
- private ControllerMetrics _controllerMetrics;
- private final LeadControllerManager _leadControllerManager;
- private final ControllerConf _controllerConf;
-
- public PinotRealtimeSegmentManager(PinotHelixResourceManager pinotManager,
- LeadControllerManager leadControllerManager, ControllerConf controllerConf) {
- _pinotHelixResourceManager = pinotManager;
- _leadControllerManager = leadControllerManager;
- _controllerConf = controllerConf;
- String clusterName = _pinotHelixResourceManager.getHelixClusterName();
- _propertyStorePath = PropertyPathBuilder.propertyStore(clusterName);
- _tableConfigPath = _propertyStorePath + TABLE_CONFIG;
- }
-
- public void start(ControllerMetrics controllerMetrics) {
- _controllerMetrics = controllerMetrics;
-
- LOGGER.info("Starting realtime segments manager, adding a listener on the property store table configs path.");
- String zkUrl = _pinotHelixResourceManager.getHelixZkURL();
- int zkClientSessionTimeoutMs =
- _controllerConf.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_SESSION_TIMEOUT_MS_CONFIG,
- CommonConstants.Helix.ZkClient.DEFAULT_SESSION_TIMEOUT_MS);
- int zkClientConnectionTimeoutMs =
- _controllerConf.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_CONNECTION_TIMEOUT_MS_CONFIG,
- CommonConstants.Helix.ZkClient.DEFAULT_CONNECT_TIMEOUT_MS);
- _zkClient = new ZkClient.Builder()
- .setZkServer(zkUrl)
- .setSessionTimeout(zkClientSessionTimeoutMs)
- .setConnectionTimeout(zkClientConnectionTimeoutMs)
- .build();
- _zkClient.setZkSerializer(new ZNRecordSerializer());
- _zkClient.waitUntilConnected(zkClientConnectionTimeoutMs, TimeUnit.MILLISECONDS);
-
- // Subscribe to any data/child changes to property
- _zkClient.subscribeChildChanges(_tableConfigPath, this);
- _zkClient.subscribeDataChanges(_tableConfigPath, this);
-
- // Setup change listeners for already existing tables, if any.
- processPropertyStoreChange(_tableConfigPath);
- }
-
- public void stop() {
- LOGGER.info("Stopping realtime segments manager, stopping property store.");
- _pinotHelixResourceManager.getPropertyStore().stop();
- }
-
- private synchronized void assignRealtimeSegmentsToServerInstancesIfNecessary() {
- // Fetch current ideal state snapshot
- Map<String, IdealState> idealStateMap = new HashMap<>();
-
- for (String realtimeTableName : _pinotHelixResourceManager.getAllRealtimeTables()) {
- TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(realtimeTableName);
-
- // Table config might have already been deleted
- if (tableConfig == null) {
- continue;
- }
-
- // Skip if the current controller isn't the leader of this table
- if (!_leadControllerManager.isLeaderForTable(realtimeTableName)) {
- continue;
- }
-
- StreamConfig metadata = new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
- if (metadata.hasHighLevelConsumerType()) {
- idealStateMap.put(realtimeTableName, _pinotHelixResourceManager.getHelixAdmin()
- .getResourceIdealState(_pinotHelixResourceManager.getHelixClusterName(), realtimeTableName));
- } else {
- LOGGER.debug("Not considering table {} for realtime segment assignment", realtimeTableName);
- }
- }
-
- List<Pair<String, String>> listOfSegmentsToAddToInstances = new ArrayList<Pair<String, String>>();
-
- for (String realtimeTableName : idealStateMap.keySet()) {
- try {
- IdealState state = idealStateMap.get(realtimeTableName);
-
- // Are there any partitions?
- if (state.getPartitionSet().isEmpty()) {
- // No, this is a brand new ideal state, so we will add one new segment to every partition and replica
- List<String> instancesInResource = new ArrayList<>();
- try {
- instancesInResource
- .addAll(_pinotHelixResourceManager.getServerInstancesForTable(realtimeTableName, TableType.REALTIME));
- } catch (Exception e) {
- LOGGER.error("Caught exception while fetching instances for resource {}", realtimeTableName, e);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- }
-
- // Assign a new segment to all server instances
- for (String instanceId : instancesInResource) {
- InstanceZKMetadata instanceZKMetadata = _pinotHelixResourceManager.getInstanceZKMetadata(instanceId);
-
- if (instanceZKMetadata == null) {
- LOGGER.warn("Instance {} has no associated instance metadata in ZK, ignoring for segment assignment.",
- instanceId);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- continue;
- }
-
- String groupId = instanceZKMetadata.getGroupId(realtimeTableName);
- String partitionId = instanceZKMetadata.getPartition(realtimeTableName);
- if (groupId != null && !groupId.isEmpty() && partitionId != null && !partitionId.isEmpty()) {
- listOfSegmentsToAddToInstances.add(Pair.of(
- new HLCSegmentName(groupId, partitionId, String.valueOf(System.currentTimeMillis())).getSegmentName(),
- instanceId));
- } else {
- LOGGER.warn(
- "Instance {} has invalid groupId ({}) and/or partitionId ({}) for resource {}, ignoring for segment"
- + " assignment.", instanceId, groupId, partitionId, realtimeTableName);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- }
- }
- } else {
- // Add all server instances to the list of instances for which to assign a realtime segment
- Set<String> instancesToAssignRealtimeSegment = new HashSet<String>();
- try {
- instancesToAssignRealtimeSegment
- .addAll(_pinotHelixResourceManager.getServerInstancesForTable(realtimeTableName, TableType.REALTIME));
- } catch (Exception e) {
- LOGGER.error("Caught exception while fetching instances for resource {}", realtimeTableName, e);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- }
-
- // Remove server instances that are currently processing a segment
- for (String partition : state.getPartitionSet()) {
- // Helix partition is the segment name
- if (SegmentName.isHighLevelConsumerSegmentName(partition)) {
- SegmentZKMetadata segmentZKMetadata = ZKMetadataProvider
- .getSegmentZKMetadata(_pinotHelixResourceManager.getPropertyStore(), realtimeTableName, partition);
- if (segmentZKMetadata == null) {
- // Segment was deleted by retention manager.
- continue;
- }
- if (segmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
- instancesToAssignRealtimeSegment.removeAll(state.getInstanceSet(partition));
- }
- }
- }
-
- // Assign a new segment to the server instances not currently processing this segment
- for (String instanceId : instancesToAssignRealtimeSegment) {
- InstanceZKMetadata instanceZKMetadata = _pinotHelixResourceManager.getInstanceZKMetadata(instanceId);
- String groupId = instanceZKMetadata.getGroupId(realtimeTableName);
- String partitionId = instanceZKMetadata.getPartition(realtimeTableName);
- listOfSegmentsToAddToInstances.add(Pair.of(
- new HLCSegmentName(groupId, partitionId, String.valueOf(System.currentTimeMillis())).getSegmentName(),
- instanceId));
- }
- }
- } catch (Exception e) {
- LOGGER.warn("Caught exception while processing resource {}, skipping.", realtimeTableName, e);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- }
- }
-
- LOGGER.info("Computed list of new segments to add : " + Arrays.toString(listOfSegmentsToAddToInstances.toArray()));
-
- // Add the new segments to the server instances
- for (final Pair<String, String> segmentIdAndInstanceId : listOfSegmentsToAddToInstances) {
- final String segmentId = segmentIdAndInstanceId.getLeft();
- final String instanceName = segmentIdAndInstanceId.getRight();
-
- try {
- final HLCSegmentName segName = new HLCSegmentName(segmentId);
- String resourceName = segName.getTableName();
-
- // Does the ideal state already contain this segment?
- if (!idealStateMap.get(resourceName).getPartitionSet().contains(segmentId)) {
- // No, add it
- // Create the realtime segment metadata
- SegmentZKMetadata segmentMetadataToAdd = new SegmentZKMetadata(segmentId);
- segmentMetadataToAdd.setStatus(Status.IN_PROGRESS);
-
- // Add the new metadata to the property store
- ZKMetadataProvider
- .setSegmentZKMetadata(_pinotHelixResourceManager.getPropertyStore(), resourceName, segmentMetadataToAdd);
-
- // Update the ideal state to add the new realtime segment
- HelixHelper.updateIdealState(_pinotHelixResourceManager.getHelixZkManager(), resourceName,
- new Function<IdealState, IdealState>() {
- @Override
- public IdealState apply(IdealState idealState) {
- return PinotTableIdealStateBuilder
- .addNewRealtimeSegmentToIdealState(segmentId, idealState, instanceName);
- }
- }, RetryPolicies.exponentialBackoffRetryPolicy(5, 500L, 2.0f));
- }
- } catch (Exception e) {
- LOGGER.warn("Caught exception while processing segment {} for instance {}, skipping.", segmentId, instanceName,
- e);
- _controllerMetrics
- .addMeteredGlobalValue(ControllerMeter.CONTROLLER_REALTIME_TABLE_SEGMENT_ASSIGNMENT_ERROR, 1L);
- }
- }
- }
-
- @Override
- public synchronized void onDataChange(String path) {
- LOGGER.info("PinotRealtimeSegmentManager.onDataChange: {}", path);
- processPropertyStoreChange(path);
- }
-
- @Override
- public synchronized void onDataCreate(String path) {
- LOGGER.info("PinotRealtimeSegmentManager.onDataCreate: {}", path);
- processPropertyStoreChange(path);
- }
-
- @Override
- public synchronized void onDataDelete(String path) {
- LOGGER.info("PinotRealtimeSegmentManager.onDataDelete: {}", path);
- processPropertyStoreChange(path);
- }
-
- private void processPropertyStoreChange(String path) {
- try {
- LOGGER.info("Processing change notification for path: {}", path);
- refreshWatchers(path);
-
- if (path.matches(REALTIME_SEGMENT_PROPERTY_STORE_PATH_PATTERN) || path
- .matches(REALTIME_TABLE_CONFIG_PROPERTY_STORE_PATH_PATTERN) || path.equals(CONTROLLER_LEADER_CHANGE)) {
- assignRealtimeSegmentsToServerInstancesIfNecessary();
- }
- } catch (Exception e) {
- LOGGER.error("Caught exception while processing change for path {}", path, e);
- Utils.rethrowException(e);
- }
- }
-
- /**
- * Helper method to perform idempotent operation to refresh all watches (related to real-time segments):
- * - Data change listener for all existing real-time tables.
- * - Child creation listener for all existing real-time tables.
- * - Data change listener for all existing real-time segments
- *
- * @param path
- */
- private void refreshWatchers(String path) {
- LOGGER.info("Received change notification for path: {}", path);
- List<Stat> stats = new ArrayList<>();
- List<ZNRecord> tableConfigs = _pinotHelixResourceManager.getPropertyStore().getChildren(TABLE_CONFIG, stats, 0);
-
- if (tableConfigs == null) {
- return;
- }
-
- for (ZNRecord tableConfigZnRecord : tableConfigs) {
- // NOTE: it is possible that znRecord is null if the record gets removed while calling this method
- if (tableConfigZnRecord == null) {
- continue;
- }
- try {
- String znRecordId = tableConfigZnRecord.getId();
- if (TableNameBuilder.getTableTypeFromTableName(znRecordId) == TableType.REALTIME) {
- TableConfig tableConfig = TableConfigUtils.fromZNRecord(tableConfigZnRecord);
- StreamConfig metadata =
- new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
- if (metadata.hasHighLevelConsumerType()) {
- String realtimeTableName = tableConfig.getTableName();
- String realtimeSegmentsPathForTable = _propertyStorePath + SEGMENTS_PATH + "/" + realtimeTableName;
-
- LOGGER.info("Setting data/child changes watch for real-time table '{}'", realtimeTableName);
- _zkClient.subscribeDataChanges(realtimeSegmentsPathForTable, this);
- _zkClient.subscribeChildChanges(realtimeSegmentsPathForTable, this);
-
- List<String> childNames =
- _pinotHelixResourceManager.getPropertyStore().getChildNames(SEGMENTS_PATH + "/" + realtimeTableName, 0);
-
- if (childNames != null && !childNames.isEmpty()) {
- for (String segmentName : childNames) {
- if (!SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
- continue;
- }
- String segmentPath = realtimeSegmentsPathForTable + "/" + segmentName;
- SegmentZKMetadata segmentZKMetadata = ZKMetadataProvider
- .getSegmentZKMetadata(_pinotHelixResourceManager.getPropertyStore(), realtimeTableName,
- segmentName);
- if (segmentZKMetadata == null) {
- // The segment got deleted by retention manager
- continue;
- }
- if (segmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
- LOGGER.info("Setting data change watch for real-time segment currently being consumed: {}",
- segmentPath);
- _zkClient.subscribeDataChanges(segmentPath, this);
- } else {
- _zkClient.unsubscribeDataChanges(segmentPath, this);
- }
- }
- }
- }
- }
- } catch (Exception e) {
- // we want to continue setting watches for other tables for any kind of exception here so that
- // errors with one table don't impact others
- LOGGER.error("Caught exception while processing ZNRecord id: {}. Skipping node to continue setting watches",
- tableConfigZnRecord.getId(), e);
- }
- }
- }
-
- @Override
- public void handleChildChange(String parentPath, List<String> currentChildren)
- throws Exception {
- LOGGER.info("PinotRealtimeSegmentManager.handleChildChange: {}", parentPath);
- processPropertyStoreChange(parentPath);
-
- // If parent path get removed, currentChildren will be null
- if (currentChildren != null) {
- for (String table : currentChildren) {
- if (table.endsWith("_REALTIME")) {
- LOGGER.info("PinotRealtimeSegmentManager.handleChildChange with table: {}", parentPath + "/" + table);
- processPropertyStoreChange(parentPath + "/" + table);
- }
- }
- }
- }
-
- @Override
- public void handleDataChange(String dataPath, Object data)
- throws Exception {
- LOGGER.info("PinotRealtimeSegmentManager.handleDataChange: {}", dataPath);
- processPropertyStoreChange(dataPath);
- }
-
- @Override
- public void handleDataDeleted(String dataPath)
- throws Exception {
- LOGGER.info("PinotRealtimeSegmentManager.handleDataDeleted: {}", dataPath);
- processPropertyStoreChange(dataPath);
- }
-}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
index 7f5f7185a9..11134f342f 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/SegmentCompletionManager.java
@@ -37,7 +37,7 @@ import org.apache.pinot.common.utils.URIUtils;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
@@ -119,10 +119,10 @@ public class SegmentCompletionManager {
}
protected StreamPartitionMsgOffsetFactory getStreamPartitionMsgOffsetFactory(LLCSegmentName llcSegmentName) {
- final String rawTableName = llcSegmentName.getTableName();
+ String rawTableName = llcSegmentName.getTableName();
TableConfig tableConfig = _segmentManager.getTableConfig(TableNameBuilder.REALTIME.tableNameWithType(rawTableName));
- PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ StreamConfig streamConfig =
+ new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
return StreamConsumerFactoryProvider.create(streamConfig).createStreamMsgOffsetFactory();
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/DefaultFlushThresholdUpdater.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/DefaultFlushThresholdUpdater.java
index 9c1ddea8e0..e1dc7f9a80 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/DefaultFlushThresholdUpdater.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/DefaultFlushThresholdUpdater.java
@@ -23,7 +23,7 @@ import java.util.List;
import javax.annotation.Nullable;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
/**
@@ -38,10 +38,9 @@ public class DefaultFlushThresholdUpdater implements FlushThresholdUpdater {
}
@Override
- public void updateFlushThreshold(PartitionLevelStreamConfig streamConfig,
- SegmentZKMetadata newSegmentZKMetadata, CommittingSegmentDescriptor committingSegmentDescriptor,
- @Nullable SegmentZKMetadata committingSegmentZKMetadata, int maxNumPartitionsPerInstance,
- List<PartitionGroupMetadata> partitionGroupMetadataList) {
+ public void updateFlushThreshold(StreamConfig streamConfig, SegmentZKMetadata newSegmentZKMetadata,
+ CommittingSegmentDescriptor committingSegmentDescriptor, @Nullable SegmentZKMetadata committingSegmentZKMetadata,
+ int maxNumPartitionsPerInstance, List<PartitionGroupMetadata> partitionGroupMetadataList) {
// Configure the segment size flush limit based on the maximum number of partitions allocated to an instance
newSegmentZKMetadata.setSizeThresholdToFlushSegment(_tableFlushSize / maxNumPartitionsPerInstance);
}
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 a166301911..318367e4d3 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
@@ -20,7 +20,7 @@ package org.apache.pinot.controller.helix.core.realtime.segment;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
/**
@@ -36,7 +36,7 @@ public class FlushThresholdUpdateManager {
* If flush size <= 0, create new SegmentSizeBasedFlushThresholdUpdater if not already created. Create only 1 per
* table because we want to maintain tuning information for the table in the updater.
*/
- public FlushThresholdUpdater getFlushThresholdUpdater(PartitionLevelStreamConfig streamConfig) {
+ public FlushThresholdUpdater getFlushThresholdUpdater(StreamConfig streamConfig) {
String realtimeTableName = streamConfig.getTableNameWithType();
int flushThresholdRows = streamConfig.getFlushThresholdRows();
@@ -44,8 +44,8 @@ public class FlushThresholdUpdateManager {
_flushThresholdUpdaterMap.remove(realtimeTableName);
return new DefaultFlushThresholdUpdater(flushThresholdRows);
} else {
- return _flushThresholdUpdaterMap
- .computeIfAbsent(realtimeTableName, k -> new SegmentSizeBasedFlushThresholdUpdater());
+ return _flushThresholdUpdaterMap.computeIfAbsent(realtimeTableName,
+ k -> new SegmentSizeBasedFlushThresholdUpdater());
}
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdater.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdater.java
index c10cbab7eb..5022ca398a 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdater.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/FlushThresholdUpdater.java
@@ -22,7 +22,7 @@ import java.util.List;
import javax.annotation.Nullable;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
/**
@@ -34,8 +34,7 @@ public interface FlushThresholdUpdater {
/**
* Updates the flush threshold for the given segment ZK metadata
*/
- void updateFlushThreshold(PartitionLevelStreamConfig streamConfig, SegmentZKMetadata newSegmentZKMetadata,
- CommittingSegmentDescriptor committingSegmentDescriptor,
- @Nullable SegmentZKMetadata committingSegmentZKMetadata, int maxNumPartitionsPerInstance,
- List<PartitionGroupMetadata> partitionGroupMetadataList);
+ void updateFlushThreshold(StreamConfig streamConfig, SegmentZKMetadata newSegmentZKMetadata,
+ CommittingSegmentDescriptor committingSegmentDescriptor, @Nullable SegmentZKMetadata committingSegmentZKMetadata,
+ int maxNumPartitionsPerInstance, List<PartitionGroupMetadata> partitionGroupMetadataList);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputer.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputer.java
index 625785d254..aba18df887 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputer.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputer.java
@@ -25,7 +25,7 @@ import javax.annotation.Nullable;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.TimeUtils;
class SegmentFlushThresholdComputer {
@@ -57,10 +57,9 @@ class SegmentFlushThresholdComputer {
return _latestSegmentRowsToSizeRatio;
}
- public int computeThreshold(PartitionLevelStreamConfig streamConfig,
- CommittingSegmentDescriptor committingSegmentDescriptor,
- @Nullable SegmentZKMetadata committingSegmentZKMetadata,
- List<PartitionGroupMetadata> partitionGroupMetadataList, String newSegmentName) {
+ public int computeThreshold(StreamConfig streamConfig, CommittingSegmentDescriptor committingSegmentDescriptor,
+ @Nullable SegmentZKMetadata committingSegmentZKMetadata, List<PartitionGroupMetadata> partitionGroupMetadataList,
+ String newSegmentName) {
final long desiredSegmentSizeBytes = streamConfig.getFlushThresholdSegmentSizeBytes();
final long optimalSegmentSizeBytesMin = desiredSegmentSizeBytes / 2;
final double optimalSegmentSizeBytesMax = desiredSegmentSizeBytes * 1.5;
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 4cb56bb510..439279afbd 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
@@ -23,7 +23,7 @@ import java.util.List;
import javax.annotation.Nullable;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,14 +51,12 @@ public class SegmentSizeBasedFlushThresholdUpdater implements FlushThresholdUpda
// synchronized since this method could be called for multiple partitions of the same table in different threads
@Override
- public synchronized void updateFlushThreshold(PartitionLevelStreamConfig streamConfig,
- SegmentZKMetadata newSegmentZKMetadata, CommittingSegmentDescriptor committingSegmentDescriptor,
- @Nullable SegmentZKMetadata committingSegmentZKMetadata, int maxNumPartitionsPerInstance,
- List<PartitionGroupMetadata> partitionGroupMetadataList) {
-
- int threshold = _flushThresholdComputer.computeThreshold(streamConfig, committingSegmentDescriptor,
- committingSegmentZKMetadata, partitionGroupMetadataList, newSegmentZKMetadata.getSegmentName());
-
+ public synchronized void updateFlushThreshold(StreamConfig streamConfig, SegmentZKMetadata newSegmentZKMetadata,
+ CommittingSegmentDescriptor committingSegmentDescriptor, @Nullable SegmentZKMetadata committingSegmentZKMetadata,
+ int maxNumPartitionsPerInstance, List<PartitionGroupMetadata> partitionGroupMetadataList) {
+ int threshold =
+ _flushThresholdComputer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ partitionGroupMetadataList, newSegmentZKMetadata.getSegmentName());
newSegmentZKMetadata.setSizeThresholdToFlushSegment(threshold);
}
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
index 88700e427b..128d71a373 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/rebalance/TableRebalancer.java
@@ -62,9 +62,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.TierConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -177,24 +175,6 @@ public class TableRebalancer {
reassignInstances, includeConsuming, bootstrap, downtime, minReplicasToKeepUpForNoDowntime,
enableStrictReplicaGroup, bestEfforts, externalViewCheckIntervalInMs, externalViewStabilizationTimeoutInMs);
- // Validate table config
- try {
- // Do not allow rebalancing HLC real-time table
- if (tableConfig.getTableType() == TableType.REALTIME && new StreamConfig(tableNameWithType,
- IngestionConfigUtils.getStreamConfigMap(tableConfig)).hasHighLevelConsumerType()) {
- LOGGER.warn("For rebalanceId: {}, cannot rebalance table: {} with high-level consumer, aborting the rebalance",
- rebalanceJobId, tableNameWithType);
- return new RebalanceResult(rebalanceJobId, RebalanceResult.Status.FAILED,
- "Cannot rebalance table with high-level consumer", null, null, null);
- }
- } catch (Exception e) {
- LOGGER.warn(
- "For rebalanceId: {}, caught exception while validating table config for table: {}, aborting the rebalance",
- rebalanceJobId, tableNameWithType, e);
- return new RebalanceResult(rebalanceJobId, RebalanceResult.Status.FAILED,
- "Caught exception while validating table config: " + e, null, null, null);
- }
-
// Fetch ideal state
PropertyKey idealStatePropertyKey = _helixDataAccessor.keyBuilder().idealStates(tableNameWithType);
IdealState currentIdealState;
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
index e0ad80bcec..ad9cd1f0f5 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/relocation/SegmentRelocator.java
@@ -48,10 +48,8 @@ import org.apache.pinot.controller.helix.core.rebalance.RebalanceResult;
import org.apache.pinot.controller.helix.core.rebalance.TableRebalancer;
import org.apache.pinot.controller.util.TableTierReader;
import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.utils.RebalanceConfigConstants;
-import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -152,19 +150,13 @@ public class SegmentRelocator extends ControllerPeriodicTask<Void> {
TableConfig tableConfig = _pinotHelixResourceManager.getTableConfig(tableNameWithType);
Preconditions.checkState(tableConfig != null, "Failed to find table config for table: {}", tableNameWithType);
- // Segment relocation doesn't apply to HLC
- boolean isRealtimeTable = TableNameBuilder.isRealtimeTableResource(tableNameWithType);
- if (isRealtimeTable && new StreamConfig(tableNameWithType,
- IngestionConfigUtils.getStreamConfigMap(tableConfig)).hasHighLevelConsumerType()) {
- return;
- }
-
boolean relocate = false;
if (TierConfigUtils.shouldRelocateToTiers(tableConfig)) {
relocate = true;
LOGGER.info("Relocating segments to tiers for table: {}", tableNameWithType);
}
- if (isRealtimeTable && InstanceAssignmentConfigUtils.shouldRelocateCompletedSegments(tableConfig)) {
+ if (tableConfig.getTableType() == TableType.REALTIME
+ && InstanceAssignmentConfigUtils.shouldRelocateCompletedSegments(tableConfig)) {
relocate = true;
LOGGER.info("Relocating COMPLETED segments for table: {}", tableNameWithType);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
index 93c7d8573b..dac61a3096 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/retention/RetentionManager.java
@@ -30,7 +30,6 @@ import org.apache.pinot.common.lineage.SegmentLineage;
import org.apache.pinot.common.lineage.SegmentLineageAccessHelper;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ControllerMetrics;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
@@ -144,13 +143,10 @@ public class RetentionManager extends ControllerPeriodicTask<Void> {
for (SegmentZKMetadata segmentZKMetadata : _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName)) {
String segmentName = segmentZKMetadata.getSegmentName();
if (segmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
- // In progress segment, only check LLC segment
- if (SegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- // Delete old LLC segment that hangs around. Do not delete segment that are current since there may be a race
- // with RealtimeSegmentValidationManager trying to auto-create the LLC segment
- if (shouldDeleteInProgressLLCSegment(segmentName, idealState, segmentZKMetadata)) {
- segmentsToDelete.add(segmentName);
- }
+ // Delete old LLC segment that hangs around. Do not delete segment that are current since there may be a race
+ // with RealtimeSegmentValidationManager trying to auto-create the LLC segment
+ if (shouldDeleteInProgressLLCSegment(segmentName, idealState, segmentZKMetadata)) {
+ segmentsToDelete.add(segmentName);
}
} else {
// Sealed segment
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/ZKMetadataUtils.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/ZKMetadataUtils.java
index 306d93d0ae..d481b669a4 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/ZKMetadataUtils.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/ZKMetadataUtils.java
@@ -26,7 +26,7 @@ import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.utils.SegmentName;
+import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.SegmentMetadata;
import org.apache.pinot.segment.spi.partition.PartitionFunction;
@@ -136,7 +136,7 @@ public class ZKMetadataUtils {
segmentZKMetadata.setStatus(CommonConstants.Segment.Realtime.Status.UPLOADED);
// For new segment, start/end offset must exist if the segment name follows LLC segment name convention
- if (newSegment && SegmentName.isLowLevelConsumerSegmentName(segmentMetadata.getName())) {
+ if (newSegment && LLCSegmentName.isLLCSegment(segmentMetadata.getName())) {
Preconditions.checkArgument(segmentMetadata.getStartOffset() != null && segmentMetadata.getEndOffset() != null,
"New uploaded LLC segment must have start/end offset in the segment metadata");
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
index 90c23361d1..d0a7bb1a37 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/validation/RealtimeSegmentValidationManager.java
@@ -26,8 +26,6 @@ import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ControllerMetrics;
import org.apache.pinot.common.metrics.ValidationMetrics;
-import org.apache.pinot.common.utils.HLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.LeadControllerManager;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
@@ -35,7 +33,7 @@ import org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTas
import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.stream.OffsetCriteria;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.slf4j.Logger;
@@ -102,32 +100,24 @@ public class RealtimeSegmentValidationManager extends ControllerPeriodicTask<Rea
LOGGER.warn("Failed to find table config for table: {}, skipping validation", tableNameWithType);
return;
}
- PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig));
-
+ StreamConfig streamConfig =
+ new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
if (context._runSegmentLevelValidation) {
runSegmentLevelValidation(tableConfig, streamConfig);
}
-
- if (streamConfig.hasLowLevelConsumerType()) {
- _llcRealtimeSegmentManager.ensureAllPartitionsConsuming(tableConfig, streamConfig,
- context._recreateDeletedConsumingSegment, context._offsetCriteria);
- }
+ _llcRealtimeSegmentManager.ensureAllPartitionsConsuming(tableConfig, streamConfig,
+ context._recreateDeletedConsumingSegment, context._offsetCriteria);
}
- private void runSegmentLevelValidation(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig) {
+ private void runSegmentLevelValidation(TableConfig tableConfig, StreamConfig streamConfig) {
String realtimeTableName = tableConfig.getTableName();
List<SegmentZKMetadata> segmentsZKMetadata = _pinotHelixResourceManager.getSegmentsZKMetadata(realtimeTableName);
// Update the total document count gauge
- // Count HLC segments if high level consumer is configured
- boolean countHLCSegments = streamConfig.hasHighLevelConsumerType();
- _validationMetrics.updateTotalDocumentCountGauge(realtimeTableName,
- computeTotalDocumentCount(segmentsZKMetadata, countHLCSegments));
+ _validationMetrics.updateTotalDocumentCountGauge(realtimeTableName, computeTotalDocumentCount(segmentsZKMetadata));
// Check missing segments and upload them to the deep store
- if (streamConfig.hasLowLevelConsumerType()
- && _llcRealtimeSegmentManager.isDeepStoreLLCSegmentUploadRetryEnabled()) {
+ if (_llcRealtimeSegmentManager.isDeepStoreLLCSegmentUploadRetryEnabled()) {
_llcRealtimeSegmentManager.uploadToDeepStoreIfMissing(tableConfig, segmentsZKMetadata);
}
}
@@ -142,34 +132,10 @@ public class RealtimeSegmentValidationManager extends ControllerPeriodicTask<Rea
}
@VisibleForTesting
- static long computeTotalDocumentCount(List<SegmentZKMetadata> segmentsZKMetadata, boolean countHLCSegments) {
+ static long computeTotalDocumentCount(List<SegmentZKMetadata> segmentsZKMetadata) {
long numTotalDocs = 0;
- if (countHLCSegments) {
- String groupId = null;
- for (SegmentZKMetadata segmentZKMetadata : segmentsZKMetadata) {
- String segmentName = segmentZKMetadata.getSegmentName();
- if (SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
- HLCSegmentName hlcSegmentName = new HLCSegmentName(segmentName);
- String segmentGroupId = hlcSegmentName.getGroupId();
- if (groupId == null) {
- groupId = segmentGroupId;
- numTotalDocs = segmentZKMetadata.getTotalDocs();
- } else {
- // Discard all segments with different group id as they are replicas
- if (groupId.equals(segmentGroupId)) {
- numTotalDocs += segmentZKMetadata.getTotalDocs();
- }
- }
- }
- }
- } else {
- for (SegmentZKMetadata segmentZKMetadata : segmentsZKMetadata) {
- String segmentName = segmentZKMetadata.getSegmentName();
- if (!SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
- // LLC segments or uploaded segments
- numTotalDocs += segmentZKMetadata.getTotalDocs();
- }
- }
+ for (SegmentZKMetadata segmentZKMetadata : segmentsZKMetadata) {
+ numTotalDocs += segmentZKMetadata.getTotalDocs();
}
return numTotalDocs;
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceAssignmentRestletResourceStatelessTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceAssignmentRestletResourceStatelessTest.java
index 570c59438c..dedc79384e 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceAssignmentRestletResourceStatelessTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/PinotInstanceAssignmentRestletResourceStatelessTest.java
@@ -95,7 +95,7 @@ public class PinotInstanceAssignmentRestletResourceStatelessTest extends Control
_helixResourceManager.addTable(offlineTableConfig);
TableConfig realtimeTableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setBrokerTenant(BROKER_TENANT_NAME)
- .setServerTenant(SERVER_TENANT_NAME).setLLC(true)
+ .setServerTenant(SERVER_TENANT_NAME)
.setStreamConfigs(FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap()).build();
_helixResourceManager.addTable(realtimeTableConfig);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
index 2ee5bb1361..25b97606fe 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/TableConfigsRestletResourceTest.java
@@ -67,8 +67,7 @@ public class TableConfigsRestletResourceTest extends ControllerTest {
} else {
StreamConfig streamConfig = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs();
return new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName("timeColumn")
- .setRetentionTimeUnit("DAYS").setLLC(true).setRetentionTimeValue("5")
- .setStreamConfigs(streamConfig.getStreamConfigsMap());
+ .setRetentionTimeUnit("DAYS").setRetentionTimeValue("5").setStreamConfigs(streamConfig.getStreamConfigsMap());
}
}
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/ZKOperatorTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/ZKOperatorTest.java
index 89adf60096..0e69dd105e 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/ZKOperatorTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/api/upload/ZKOperatorTest.java
@@ -89,7 +89,7 @@ public class ZKOperatorTest {
TableConfig offlineTableConfig = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME).build();
TableConfig realtimeTableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setTimeColumnName(TIME_COLUMN)
- .setStreamConfigs(getStreamConfigs()).setLLC(true).setNumReplicas(1).build();
+ .setStreamConfigs(getStreamConfigs()).setNumReplicas(1).build();
_resourceManager.addSchema(schema, false, false);
_resourceManager.addTable(offlineTableConfig);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
index 31682d911e..28fd0ec795 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
@@ -63,7 +63,7 @@ public class RealtimeConsumerMonitorTest {
List<String> allTableNames = new ArrayList<String>();
allTableNames.add(tableName);
TableConfig tableConfig =
- new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName("timeColumn").setLLC(true)
+ new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName("timeColumn")
.setNumReplicas(2).setStreamConfigs(getStreamConfigMap()).build();
LLCSegmentName segmentPartition1Seq0 = new LLCSegmentName(rawTableName, 1, 0, System.currentTimeMillis());
LLCSegmentName segmentPartition1Seq1 = new LLCSegmentName(rawTableName, 1, 1, System.currentTimeMillis());
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
index 22ddedd257..731f1f33d5 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/SegmentStatusCheckerTest.java
@@ -55,7 +55,10 @@ import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.testng.Assert;
import org.testng.annotations.Test;
-import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -178,7 +181,7 @@ public class SegmentStatusCheckerTest {
List<String> allTableNames = new ArrayList<String>();
allTableNames.add(tableName);
TableConfig tableConfig =
- new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName("timeColumn").setLLC(true)
+ new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setTimeColumnName("timeColumn")
.setNumReplicas(3).setStreamConfigs(getStreamConfigMap()).build();
final LLCSegmentName seg1 = new LLCSegmentName(rawTableName, 1, 0, System.currentTimeMillis());
final LLCSegmentName seg2 = new LLCSegmentName(rawTableName, 1, 1, System.currentTimeMillis());
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupSegmentAssignmentTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupSegmentAssignmentTest.java
index 2bf6fd11e0..8bfa93a1aa 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupSegmentAssignmentTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupSegmentAssignmentTest.java
@@ -83,7 +83,7 @@ public class RealtimeNonReplicaGroupSegmentAssignmentTest {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
TableConfig tableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
- .setLLC(true).setStreamConfigs(streamConfigs).build();
+ .setStreamConfigs(streamConfigs).build();
_segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(createHelixManager(), tableConfig);
_instancePartitionsMap = new TreeMap<>();
@@ -118,7 +118,7 @@ public class RealtimeNonReplicaGroupSegmentAssignmentTest {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
TableConfig tableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
- .setLLC(true).setStreamConfigs(streamConfigs).build();
+ .setStreamConfigs(streamConfigs).build();
// Update the replication by changing the NUM_REPLICAS_PER_PARTITION
tableConfig.getValidationConfig().setReplicasPerPartition(NUM_REPLICAS_PER_PARTITION);
SegmentAssignment segmentAssignment =
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupTieredSegmentAssignmentTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupTieredSegmentAssignmentTest.java
index 73cb94c4fa..2686f08178 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupTieredSegmentAssignmentTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeNonReplicaGroupTieredSegmentAssignmentTest.java
@@ -122,7 +122,7 @@ public class RealtimeNonReplicaGroupTieredSegmentAssignmentTest {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
TableConfig tableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
- .setTierConfigList(tierConfigList).setLLC(true).setStreamConfigs(streamConfigs).build();
+ .setTierConfigList(tierConfigList).setStreamConfigs(streamConfigs).build();
_segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(null, tableConfig);
_instancePartitionsMap = new TreeMap<>();
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
index 713b4c442a..ffc9998179 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeReplicaGroupSegmentAssignmentTest.java
@@ -87,7 +87,7 @@ public class RealtimeReplicaGroupSegmentAssignmentTest {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
TableConfig tableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(NUM_REPLICAS)
- .setLLC(true).setStreamConfigs(streamConfigs)
+ .setStreamConfigs(streamConfigs)
.setSegmentAssignmentStrategy(AssignmentStrategy.REPLICA_GROUP_SEGMENT_ASSIGNMENT_STRATEGY)
.setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig(PARTITION_COLUMN, 1)).build();
_segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(createHelixManager(), tableConfig);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
index 619d61ef82..e452820c98 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/assignment/segment/strategy/SegmentAssignmentStrategyFactoryTest.java
@@ -100,8 +100,8 @@ public class SegmentAssignmentStrategyFactoryTest {
@Test
public void testBalancedNumSegmentAssignmentStrategyForRealtimeTables() {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
- TableConfig tableConfig = new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setLLC(true)
- .setStreamConfigs(streamConfigs).build();
+ TableConfig tableConfig =
+ new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setStreamConfigs(streamConfigs).build();
InstancePartitions instancePartitions = new InstancePartitions(INSTANCE_PARTITIONS_NAME);
instancePartitions.setInstances(0, 0, INSTANCES);
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 2223c9e341..d367741080 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
@@ -70,7 +70,6 @@ import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Helix;
@@ -90,12 +89,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-import static org.testng.Assert.assertFalse;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
-import static org.testng.Assert.assertTrue;
-import static org.testng.Assert.fail;
+import static org.testng.Assert.*;
public class PinotLLCRealtimeSegmentManagerTest {
@@ -757,20 +751,6 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
}
- @Test(expectedExceptions = IllegalStateException.class)
- public void testPreExistingSegments() {
- FakePinotLLCRealtimeSegmentManager segmentManager = new FakePinotLLCRealtimeSegmentManager();
- segmentManager._numReplicas = 2;
- segmentManager.makeTableConfig();
- segmentManager._numInstances = 5;
- segmentManager.makeConsumingInstancePartitions();
- segmentManager._numPartitions = 4;
-
- String existingSegmentName = new LLCSegmentName(RAW_TABLE_NAME, 0, 0, CURRENT_TIME_MS).getSegmentName();
- segmentManager._segmentZKMetadataMap.put(existingSegmentName, new SegmentZKMetadata(existingSegmentName));
- segmentManager.setUpNewTable();
- }
-
@Test
public void testCommitSegmentWhenControllerWentThroughGC() {
// Set up a new table with 2 replicas, 5 instances, 4 partitions
@@ -868,12 +848,6 @@ public class PinotLLCRealtimeSegmentManagerTest {
} catch (IllegalStateException e) {
// Expected
}
- try {
- segmentManager.removeLLCSegments(new IdealState(REALTIME_TABLE_NAME));
- fail();
- } catch (IllegalStateException e) {
- // Expected
- }
try {
segmentManager.commitSegmentFile(REALTIME_TABLE_NAME, mock(CommittingSegmentDescriptor.class));
fail();
@@ -1089,7 +1063,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
int _numReplicas;
TableConfig _tableConfig;
- PartitionLevelStreamConfig _streamConfig;
+ StreamConfig _streamConfig;
int _numInstances;
InstancePartitions _consumingInstancePartitions;
Map<String, SegmentZKMetadata> _segmentZKMetadataMap = new HashMap<>();
@@ -1112,9 +1086,9 @@ public class PinotLLCRealtimeSegmentManagerTest {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
_tableConfig =
new TableConfigBuilder(TableType.REALTIME).setTableName(RAW_TABLE_NAME).setNumReplicas(_numReplicas)
- .setLLC(true).setStreamConfigs(streamConfigs).build();
- _streamConfig = new PartitionLevelStreamConfig(_tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(_tableConfig));
+ .setStreamConfigs(streamConfigs).build();
+ _streamConfig =
+ new StreamConfig(_tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(_tableConfig));
}
void makeConsumingInstancePartitions() {
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 0e45731f94..fbdaec5d3e 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
@@ -26,7 +26,6 @@ import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.testng.annotations.Test;
@@ -66,7 +65,7 @@ public class FlushThresholdUpdaterTest {
StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
// Flush threshold rows set to 0 - SegmentSizeBasedFlushThresholdUpdater should be returned
- PartitionLevelStreamConfig autotuneStreamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig autotuneStreamConfig = mockDefaultAutotuneStreamConfig();
FlushThresholdUpdater autotuneFlushThresholdUpdater =
flushThresholdUpdateManager.getFlushThresholdUpdater(autotuneStreamConfig);
assertTrue(autotuneFlushThresholdUpdater instanceof SegmentSizeBasedFlushThresholdUpdater);
@@ -92,16 +91,16 @@ public class FlushThresholdUpdaterTest {
autotuneFlushThresholdUpdater);
}
- private PartitionLevelStreamConfig mockStreamConfig(int flushThresholdRows) {
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ private StreamConfig mockStreamConfig(int flushThresholdRows) {
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getTableNameWithType()).thenReturn(REALTIME_TABLE_NAME);
when(streamConfig.getFlushThresholdRows()).thenReturn(flushThresholdRows);
return streamConfig;
}
- private PartitionLevelStreamConfig mockAutotuneStreamConfig(long flushSegmentDesiredSizeBytes,
+ private StreamConfig mockAutotuneStreamConfig(long flushSegmentDesiredSizeBytes,
long flushThresholdTimeMillis, int flushAutotuneInitialRows) {
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getTableNameWithType()).thenReturn(REALTIME_TABLE_NAME);
when(streamConfig.getFlushThresholdRows()).thenReturn(0);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(flushSegmentDesiredSizeBytes);
@@ -110,7 +109,7 @@ public class FlushThresholdUpdaterTest {
return streamConfig;
}
- private PartitionLevelStreamConfig mockDefaultAutotuneStreamConfig() {
+ private StreamConfig mockDefaultAutotuneStreamConfig() {
return mockAutotuneStreamConfig(StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES,
StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS, StreamConfig.DEFAULT_FLUSH_AUTOTUNE_INITIAL_ROWS);
}
@@ -123,7 +122,7 @@ public class FlushThresholdUpdaterTest {
*/
@Test
public void testSegmentSizeBasedFlushThreshold() {
- PartitionLevelStreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
long desiredSegmentSizeBytes = streamConfig.getFlushThresholdSegmentSizeBytes();
long segmentSizeLowerLimit = (long) (desiredSegmentSizeBytes * 0.99);
long segmentSizeHigherLimit = (long) (desiredSegmentSizeBytes * 1.01);
@@ -161,7 +160,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testSegmentSizeBasedFlushThresholdMinPartition() {
- PartitionLevelStreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
long desiredSegmentSizeBytes = streamConfig.getFlushThresholdSegmentSizeBytes();
long segmentSizeLowerLimit = (long) (desiredSegmentSizeBytes * 0.99);
long segmentSizeHigherLimit = (long) (desiredSegmentSizeBytes * 1.01);
@@ -240,7 +239,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testTimeThreshold() {
SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
- PartitionLevelStreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -273,7 +272,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testMinThreshold() {
SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
- PartitionLevelStreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
// Start consumption
SegmentZKMetadata newSegmentZKMetadata = getNewSegmentZKMetadata(0);
@@ -305,7 +304,7 @@ public class FlushThresholdUpdaterTest {
@Test
public void testNonZeroPartitionUpdates() {
SegmentSizeBasedFlushThresholdUpdater flushThresholdUpdater = new SegmentSizeBasedFlushThresholdUpdater();
- PartitionLevelStreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
+ StreamConfig streamConfig = mockDefaultAutotuneStreamConfig();
// Start consumption for 2 partitions
SegmentZKMetadata newSegmentZKMetadataForPartition0 = getNewSegmentZKMetadata(0);
@@ -362,7 +361,7 @@ public class FlushThresholdUpdaterTest {
long flushSegmentDesiredSizeBytes = StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES / 2;
long flushThresholdTimeMillis = StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS / 2;
int flushAutotuneInitialRows = StreamConfig.DEFAULT_FLUSH_AUTOTUNE_INITIAL_ROWS / 2;
- PartitionLevelStreamConfig streamConfig =
+ StreamConfig streamConfig =
mockAutotuneStreamConfig(flushSegmentDesiredSizeBytes, flushThresholdTimeMillis, flushAutotuneInitialRows);
// Start consumption
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputerTest.java
index 6cb21af4ff..820b40a491 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/segment/SegmentFlushThresholdComputerTest.java
@@ -25,7 +25,7 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.testng.annotations.Test;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -40,15 +40,14 @@ public class SegmentFlushThresholdComputerTest {
int autoTuneInitialRows = 1_000;
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushAutotuneInitialRows()).thenReturn(autoTuneInitialRows);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
SegmentZKMetadata committingSegmentZKMetadata = null;
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "newSegmentName");
assertEquals(threshold, autoTuneInitialRows);
@@ -58,18 +57,17 @@ public class SegmentFlushThresholdComputerTest {
public void testUseLastSegmentSizeTimesRatioIfFirstSegmentInPartitionAndNewPartitionGroup() {
double segmentRowsToSizeRatio = 1.5;
long segmentSizeBytes = 20000L;
- SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer(
- Clock.systemUTC(), segmentRowsToSizeRatio);
+ SegmentFlushThresholdComputer computer =
+ new SegmentFlushThresholdComputer(Clock.systemUTC(), segmentRowsToSizeRatio);
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(segmentSizeBytes);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
SegmentZKMetadata committingSegmentZKMetadata = null;
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "newSegmentName");
// segmentSize * 1.5
@@ -81,18 +79,17 @@ public class SegmentFlushThresholdComputerTest {
public void testUseLastSegmentSizeTimesRatioIfFirstSegmentInPartitionAndNewPartitionGroupMinimumSize10000Rows() {
double segmentRowsToSizeRatio = 1.5;
long segmentSizeBytes = 2000L;
- SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer(
- Clock.systemUTC(), segmentRowsToSizeRatio);
+ SegmentFlushThresholdComputer computer =
+ new SegmentFlushThresholdComputer(Clock.systemUTC(), segmentRowsToSizeRatio);
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(segmentSizeBytes);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
SegmentZKMetadata committingSegmentZKMetadata = null;
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "newSegmentName");
assertEquals(threshold, 10000);
@@ -104,7 +101,7 @@ public class SegmentFlushThresholdComputerTest {
int segmentSizeThreshold = 5_000;
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdTimeMillis()).thenReturn(123L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
@@ -114,8 +111,7 @@ public class SegmentFlushThresholdComputerTest {
when(committingSegmentZKMetadata.getSizeThresholdToFlushSegment()).thenReturn(segmentSizeThreshold);
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "newSegmentName");
assertEquals(threshold, segmentSizeThreshold);
@@ -128,10 +124,9 @@ public class SegmentFlushThresholdComputerTest {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer(clock);
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
- when(streamConfig.getFlushThresholdTimeMillis()).thenReturn(
- MILLISECONDS.convert(6, TimeUnit.HOURS));
+ when(streamConfig.getFlushThresholdTimeMillis()).thenReturn(MILLISECONDS.convert(6, TimeUnit.HOURS));
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
when(committingSegmentDescriptor.getSegmentSizeBytes()).thenReturn(200_0000L);
@@ -143,8 +138,7 @@ public class SegmentFlushThresholdComputerTest {
currentTime - MILLISECONDS.convert(1, TimeUnit.HOURS));
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// totalDocs * 1.1
@@ -159,10 +153,9 @@ public class SegmentFlushThresholdComputerTest {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer(clock);
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
- when(streamConfig.getFlushThresholdTimeMillis()).thenReturn(
- MILLISECONDS.convert(1, TimeUnit.HOURS));
+ when(streamConfig.getFlushThresholdTimeMillis()).thenReturn(MILLISECONDS.convert(1, TimeUnit.HOURS));
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
when(committingSegmentDescriptor.getSegmentSizeBytes()).thenReturn(200_0000L);
@@ -174,8 +167,7 @@ public class SegmentFlushThresholdComputerTest {
currentTime - MILLISECONDS.convert(2, TimeUnit.HOURS));
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// (totalDocs / 2) * 1.1
@@ -188,7 +180,7 @@ public class SegmentFlushThresholdComputerTest {
public void testSegmentSizeTooSmall() {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
@@ -199,8 +191,7 @@ public class SegmentFlushThresholdComputerTest {
when(committingSegmentZKMetadata.getSizeThresholdToFlushSegment()).thenReturn(20_000);
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// totalDocs / 2
@@ -212,7 +203,7 @@ public class SegmentFlushThresholdComputerTest {
public void testSegmentSizeTooBig() {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(500_0000L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
@@ -223,8 +214,7 @@ public class SegmentFlushThresholdComputerTest {
when(committingSegmentZKMetadata.getSizeThresholdToFlushSegment()).thenReturn(20_000);
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// totalDocs + (totalDocs / 2)
@@ -236,7 +226,7 @@ public class SegmentFlushThresholdComputerTest {
public void testSegmentSizeJustRight() {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
@@ -247,8 +237,7 @@ public class SegmentFlushThresholdComputerTest {
when(committingSegmentZKMetadata.getSizeThresholdToFlushSegment()).thenReturn(20_000);
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// (totalDocs / segmentSize) * flushThresholdSegmentSize
@@ -260,7 +249,7 @@ public class SegmentFlushThresholdComputerTest {
public void testNoRows() {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
@@ -271,8 +260,7 @@ public class SegmentFlushThresholdComputerTest {
when(committingSegmentZKMetadata.getSizeThresholdToFlushSegment()).thenReturn(0);
List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- int threshold = computer.computeThreshold(
- streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
+ int threshold = computer.computeThreshold(streamConfig, committingSegmentDescriptor, committingSegmentZKMetadata,
partitionGroupMetadataList, "events3__0__0__20211222T1646Z");
// max((totalDocs / segmentSize) * flushThresholdSegmentSize, 10000)
@@ -284,7 +272,7 @@ public class SegmentFlushThresholdComputerTest {
public void testAdjustRowsToSizeRatio() {
SegmentFlushThresholdComputer computer = new SegmentFlushThresholdComputer();
- PartitionLevelStreamConfig streamConfig = mock(PartitionLevelStreamConfig.class);
+ StreamConfig streamConfig = mock(StreamConfig.class);
when(streamConfig.getFlushThresholdSegmentSizeBytes()).thenReturn(300_0000L);
CommittingSegmentDescriptor committingSegmentDescriptor = mock(CommittingSegmentDescriptor.class);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
index 2f16f830b9..46f862acba 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/retention/RetentionManagerTest.java
@@ -154,9 +154,8 @@ public class RetentionManagerTest {
private TableConfig createRealtimeTableConfig1(int replicaCount) {
Map<String, String> streamConfigs = FakeStreamConfigUtils.getDefaultLowLevelStreamConfigs().getStreamConfigsMap();
- return new TableConfigBuilder(TableType.REALTIME).setTableName(TEST_TABLE_NAME).setLLC(true)
- .setStreamConfigs(streamConfigs).setRetentionTimeUnit("DAYS").setRetentionTimeValue("5")
- .setNumReplicas(replicaCount).build();
+ return new TableConfigBuilder(TableType.REALTIME).setTableName(TEST_TABLE_NAME).setStreamConfigs(streamConfigs)
+ .setRetentionTimeUnit("DAYS").setRetentionTimeValue("5").setNumReplicas(replicaCount).build();
}
private void setupPinotHelixResourceManager(TableConfig tableConfig, final List<String> removedSegments,
@@ -273,7 +272,7 @@ public class RetentionManagerTest {
List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>();
IdealState idealState =
- PinotTableIdealStateBuilder.buildEmptyRealtimeIdealStateFor(REALTIME_TABLE_NAME, replicaCount, true);
+ PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME, replicaCount, true);
final int kafkaPartition = 5;
final long millisInDays = TimeUnit.DAYS.toMillis(1);
@@ -337,7 +336,7 @@ public class RetentionManagerTest {
List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>();
IdealState idealState =
- PinotTableIdealStateBuilder.buildEmptyRealtimeIdealStateFor(REALTIME_TABLE_NAME, replicaCount, true);
+ PinotTableIdealStateBuilder.buildEmptyIdealStateFor(REALTIME_TABLE_NAME, replicaCount, true);
final int kafkaPartition = 5;
final long millisInDays = TimeUnit.DAYS.toMillis(1);
diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
index c371dc9dc2..4d576bb8cf 100644
--- a/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
+++ b/pinot-controller/src/test/java/org/apache/pinot/controller/validation/ValidationManagerTest.java
@@ -22,7 +22,6 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.helix.model.ExternalView;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
-import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.controller.helix.ControllerTest;
import org.apache.pinot.controller.utils.SegmentMetadataMockUtils;
@@ -102,32 +101,14 @@ public class ValidationManagerTest {
@Test
public void testTotalDocumentCountRealTime() {
- // Create a bunch of dummy segments
- final String group1 = TEST_TABLE_NAME + "_REALTIME_1466446700000_34";
- final String group2 = TEST_TABLE_NAME + "_REALTIME_1466446700000_17";
- String segmentName1 = new HLCSegmentName(group1, "0", "1").getSegmentName();
- String segmentName2 = new HLCSegmentName(group1, "0", "2").getSegmentName();
- String segmentName3 = new HLCSegmentName(group1, "0", "3").getSegmentName();
- String segmentName4 = new HLCSegmentName(group2, "0", "3").getSegmentName();
-
+ // Create some dummy LLC segments (both committed and uploaded)
List<SegmentZKMetadata> segmentsZKMetadata = new ArrayList<>();
- segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName1, 10));
- segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName2, 20));
- segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName3, 30));
- // This should get ignored in the count as it belongs to a different group id
- segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName4, 20));
-
- assertEquals(RealtimeSegmentValidationManager.computeTotalDocumentCount(segmentsZKMetadata, true), 60);
-
- // Now add some LLC segments (both committed and uploaded)
String segmentName5 = new LLCSegmentName(TEST_TABLE_NAME, 1, 0, 1000).getSegmentName();
String segmentName6 = new LLCSegmentName(TEST_TABLE_NAME, 2, 27, 10000).getSegmentName();
segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName5, 10));
segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(segmentName6, 5));
segmentsZKMetadata.add(SegmentMetadataMockUtils.mockSegmentZKMetadata(TEST_SEGMENT_NAME, 15));
-
- // Only the LLC segments should get counted.
- assertEquals(RealtimeSegmentValidationManager.computeTotalDocumentCount(segmentsZKMetadata, false), 30);
+ assertEquals(RealtimeSegmentValidationManager.computeTotalDocumentCount(segmentsZKMetadata), 30);
}
@Test
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
index 002e95f1d3..ffa16052ee 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java
@@ -79,9 +79,9 @@ import org.apache.pinot.spi.stream.OffsetCriteria;
import org.apache.pinot.spi.stream.PartitionGroupConsumer;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionLagState;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.PermanentConsumerException;
import org.apache.pinot.spi.stream.RowMetadata;
+import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamDataDecoder;
@@ -280,7 +280,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
private final ServerSegmentCompletionProtocolHandler _protocolHandler;
private final long _consumeStartTime;
private final StreamPartitionMsgOffset _startOffset;
- private final PartitionLevelStreamConfig _partitionLevelStreamConfig;
+ private final StreamConfig _streamConfig;
private RowMetadata _lastRowMetadata;
private long _lastConsumedTimestampMs = -1;
@@ -403,8 +403,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
removeSegmentFile();
_numRowsErrored = 0;
- final long idlePipeSleepTimeMillis = 100;
- final long idleTimeoutMillis = _partitionLevelStreamConfig.getIdleTimeoutMillis();
+ long idlePipeSleepTimeMillis = 100;
+ long idleTimeoutMillis = _streamConfig.getIdleTimeoutMillis();
_idleTimer.init();
StreamPartitionMsgOffset lastUpdatedOffset = _streamPartitionMsgOffsetFactory
@@ -416,8 +416,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// Update _currentOffset upon return from this method
MessageBatch messageBatch;
try {
- messageBatch = _partitionGroupConsumer
- .fetchMessages(_currentOffset, null, _partitionLevelStreamConfig.getFetchTimeoutMillis());
+ messageBatch =
+ _partitionGroupConsumer.fetchMessages(_currentOffset, null, _streamConfig.getFetchTimeoutMillis());
if (_segmentLogger.isDebugEnabled()) {
_segmentLogger.debug("message batch received. filtered={} unfiltered={} endOfPartitionGroup={}",
messageBatch.getMessageCount(), messageBatch.getUnfilteredMessageCount(),
@@ -1335,11 +1335,10 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
String timeColumnName = tableConfig.getValidationConfig().getTimeColumnName();
// TODO Validate configs
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
- _partitionLevelStreamConfig =
- new PartitionLevelStreamConfig(_tableNameWithType, IngestionConfigUtils.getStreamConfigMap(_tableConfig));
- _streamConsumerFactory = StreamConsumerFactoryProvider.create(_partitionLevelStreamConfig);
+ _streamConfig = new StreamConfig(_tableNameWithType, IngestionConfigUtils.getStreamConfigMap(_tableConfig));
+ _streamConsumerFactory = StreamConsumerFactoryProvider.create(_streamConfig);
_streamPartitionMsgOffsetFactory = _streamConsumerFactory.createStreamMsgOffsetFactory();
- String streamTopic = _partitionLevelStreamConfig.getTopicName();
+ String streamTopic = _streamConfig.getTopicName();
_segmentNameStr = _segmentZKMetadata.getSegmentName();
_partitionGroupId = llcSegmentName.getPartitionGroupId();
_partitionGroupConsumptionStatus =
@@ -1358,7 +1357,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
serverMetrics);
_rateLimiter = RealtimeConsumptionRateManager.getInstance()
- .createRateLimiter(_partitionLevelStreamConfig, _tableNameWithType, _serverMetrics, _clientId);
+ .createRateLimiter(_streamConfig, _tableNameWithType, _serverMetrics, _clientId);
List<String> sortedColumns = indexLoadingConfig.getSortedColumns();
String sortedColumn;
@@ -1388,7 +1387,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
}
// Read the max number of rows
- int segmentMaxRowCount = _partitionLevelStreamConfig.getFlushThresholdRows();
+ int segmentMaxRowCount = _streamConfig.getFlushThresholdRows();
int flushThresholdSize = segmentZKMetadata.getSizeThresholdToFlushSegment();
if (flushThresholdSize > 0) {
segmentMaxRowCount = flushThresholdSize;
@@ -1429,8 +1428,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// Create message decoder
Set<String> fieldsToRead = IngestionUtils.getFieldsForRecordExtractor(_tableConfig.getIngestionConfig(), _schema);
try {
- StreamMessageDecoder streamMessageDecoder =
- StreamDecoderProvider.create(_partitionLevelStreamConfig, fieldsToRead);
+ StreamMessageDecoder streamMessageDecoder = StreamDecoderProvider.create(_streamConfig, fieldsToRead);
_streamDataDecoder = new StreamDataDecoderImpl(streamMessageDecoder);
} catch (Exception e) {
_realtimeTableDataManager.addSegmentError(_segmentNameStr,
@@ -1484,7 +1482,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
}
private void setConsumeEndTime(SegmentZKMetadata segmentZKMetadata, long now) {
- long maxConsumeTimeMillis = _partitionLevelStreamConfig.getFlushThresholdTimeMillis();
+ long maxConsumeTimeMillis = _streamConfig.getFlushThresholdTimeMillis();
_consumeEndTime = segmentZKMetadata.getCreationTime() + maxConsumeTimeMillis;
// When we restart a server, the consuming segments retain their creationTime (derived from segment
@@ -1557,9 +1555,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// However this is not an issue for Kafka, since partitionGroups never expire and every partitionGroup has
// a single partition
// Fix this before opening support for partitioning in Kinesis
- int numPartitionGroups = _partitionMetadataProvider
- .computePartitionGroupMetadata(_clientId, _partitionLevelStreamConfig,
- Collections.emptyList(), /*maxWaitTimeMs=*/5000).size();
+ int numPartitionGroups = _partitionMetadataProvider.computePartitionGroupMetadata(_clientId, _streamConfig,
+ Collections.emptyList(), /*maxWaitTimeMs=*/5000).size();
if (numPartitionGroups != numPartitions) {
_segmentLogger.warn(
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
index b963ac8bc7..3df3b91a6d 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java
@@ -42,7 +42,6 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.common.utils.SegmentUtils;
import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.common.utils.fetcher.SegmentFetcherFactory;
@@ -401,10 +400,6 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
TableConfig tableConfig = indexLoadingConfig.getTableConfig();
Schema schema = indexLoadingConfig.getSchema();
assert schema != null;
- boolean isHLCSegment = SegmentName.isHighLevelConsumerSegmentName(segmentName);
- if (isHLCSegment) {
- throw new UnsupportedOperationException("Adding high level consumer segment " + segmentName + " is not allowed");
- }
if (segmentZKMetadata.getStatus().isCompleted()) {
if (tryLoadExistingSegment(segmentName, indexLoadingConfig, segmentZKMetadata)) {
// The existing completed segment has been loaded successfully
@@ -550,8 +545,8 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
@Override
protected boolean allowDownload(String segmentName, SegmentZKMetadata zkMetadata) {
- // Cannot download HLC segment or consuming segment
- if (SegmentName.isHighLevelConsumerSegmentName(segmentName) || zkMetadata.getStatus() == Status.IN_PROGRESS) {
+ // Cannot download consuming segment
+ if (zkMetadata.getStatus() == Status.IN_PROGRESS) {
return false;
}
// TODO: may support download from peer servers as well.
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
index e865e44c5e..a2838d543f 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManagerTest.java
@@ -29,7 +29,6 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.LLCSegmentName;
import org.apache.pinot.common.utils.SchemaUtils;
import org.apache.pinot.common.utils.TarGzCompressionUtils;
@@ -214,12 +213,6 @@ public class RealtimeTableDataManagerTest {
public void testAllowDownload() {
RealtimeTableDataManager mgr = new RealtimeTableDataManager(null);
- String groupId = "myTable_REALTIME_1234567_0";
- String partitionRange = "ALL";
- String sequenceNumber = "1234567";
- HLCSegmentName hlc = new HLCSegmentName(groupId, partitionRange, sequenceNumber);
- assertFalse(mgr.allowDownload(hlc.getSegmentName(), null));
-
LLCSegmentName llc = new LLCSegmentName("tbl01", 0, 1000000, System.currentTimeMillis());
SegmentZKMetadata zkmd = mock(SegmentZKMetadata.class);
when(zkmd.getStatus()).thenReturn(Status.IN_PROGRESS);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactoryTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactoryTest.java
index 8e60b1e414..f63b91433d 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactoryTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/SegmentCommitterFactoryTest.java
@@ -52,10 +52,7 @@ public class SegmentCommitterFactoryTest {
}
private TableConfigBuilder createRealtimeTableConfig(String tableName, Map<String, String> realtimeStreamConfig) {
- return new TableConfigBuilder(TableType.REALTIME)
- .setTableName(tableName)
- .setLLC(true)
- .setStreamConfigs(realtimeStreamConfig);
+ return new TableConfigBuilder(TableType.REALTIME).setTableName(tableName).setStreamConfigs(realtimeStreamConfig);
}
@Test (description = "if controller doesn't support split commit, it should return default segment committer")
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConfigUtils.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConfigUtils.java
index 8d06b86a59..af9c4c8602 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConfigUtils.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConfigUtils.java
@@ -140,12 +140,8 @@ public class FakeStreamConfigUtils {
*/
public static StreamConfig getDefaultLowLevelStreamConfigs(int numPartitions) {
Map<String, String> streamConfigMap = getDefaultStreamConfigs();
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.LOWLEVEL.toString());
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE, NUM_PARTITIONS_KEY),
String.valueOf(numPartitions));
-
return new StreamConfig(TABLE_NAME_WITH_TYPE, streamConfigMap);
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/stream/StreamConfigTest.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/stream/StreamConfigTest.java
index 253d609d8f..333eecab04 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/stream/StreamConfigTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/stream/StreamConfigTest.java
@@ -23,14 +23,16 @@ import java.util.Map;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConsumerFactory;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamMessageDecoder;
import org.apache.pinot.spi.stream.OffsetCriteria;
-import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.DataSizeUtils;
import org.apache.pinot.spi.utils.TimeUtils;
-import org.testng.Assert;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+
public class StreamConfigTest {
@@ -39,116 +41,76 @@ public class StreamConfigTest {
*/
@Test
public void testStreamConfig() {
- boolean exception = false;
- StreamConfig streamConfig;
String streamType = "fakeStream";
String topic = "fakeTopic";
String tableName = "fakeTable_REALTIME";
- String consumerType = StreamConfig.ConsumerType.LOWLEVEL.toString();
String consumerFactoryClass = FakeStreamConsumerFactory.class.getName();
String decoderClass = FakeStreamMessageDecoder.class.getName();
// test with empty map
try {
Map<String, String> streamConfigMap = new HashMap<>();
- streamConfig = new StreamConfig(tableName, streamConfigMap);
+ new StreamConfig(tableName, streamConfigMap);
+ fail();
} catch (NullPointerException e) {
- exception = true;
+ // Expected
}
- Assert.assertTrue(exception);
// All mandatory properties set
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
- streamConfig = new StreamConfig(tableName, streamConfigMap);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
+ StreamConfig streamConfig = new StreamConfig(tableName, streamConfigMap);
+ assertEquals(streamConfig.getType(), streamType);
+ assertEquals(streamConfig.getTopicName(), topic);
+ assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
+ assertEquals(streamConfig.getDecoderClass(), decoderClass);
// Missing streamType
streamConfigMap.remove(StreamConfigProperties.STREAM_TYPE);
- exception = false;
try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
+ new StreamConfig(tableName, streamConfigMap);
+ fail();
} catch (NullPointerException e) {
- exception = true;
+ // Expected
}
- Assert.assertTrue(exception);
// Missing stream topic
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .remove(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME));
- exception = false;
- try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- } catch (NullPointerException e) {
- exception = true;
- }
- Assert.assertTrue(exception);
-
- // Missing consumer type
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
streamConfigMap.remove(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES));
- exception = false;
- try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- } catch (NullPointerException e) {
- exception = true;
- }
- Assert.assertTrue(exception);
-
- // Missing consumer factory
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.remove(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS));
- exception = false;
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME));
try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
+ new StreamConfig(tableName, streamConfigMap);
+ fail();
} catch (NullPointerException e) {
- exception = true;
+ // Expected
}
- Assert.assertFalse(exception);
- Assert.assertEquals(streamConfig.getConsumerFactoryClassName(),
- StreamConfig.DEFAULT_CONSUMER_FACTORY_CLASS_NAME_STRING);
+
+ // Missing consumer factory - allowed
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.remove(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS));
+ streamConfig = new StreamConfig(tableName, streamConfigMap);
+ assertEquals(streamConfig.getConsumerFactoryClassName(), StreamConfig.DEFAULT_CONSUMER_FACTORY_CLASS_NAME_STRING);
// Missing decoder class
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
streamConfigMap.remove(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS));
- exception = false;
try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
+ new StreamConfig(tableName, streamConfigMap);
+ fail();
} catch (NullPointerException e) {
- exception = true;
+ // Expected
}
- Assert.assertTrue(exception);
-
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getType(), streamType);
- Assert.assertEquals(streamConfig.getTopicName(), topic);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
- Assert.assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
- Assert.assertEquals(streamConfig.getDecoderClass(), decoderClass);
}
/**
@@ -158,45 +120,35 @@ public class StreamConfigTest {
public void testStreamConfigDefaults() {
String streamType = "fakeStream";
String topic = "fakeTopic";
- String consumerType = "simple";
String tableName = "fakeTable_REALTIME";
String consumerFactoryClass = FakeStreamConsumerFactory.class.getName();
String decoderClass = FakeStreamMessageDecoder.class.getName();
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
// Mandatory values + defaults
StreamConfig streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getType(), streamType);
- Assert.assertEquals(streamConfig.getTopicName(), topic);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
- Assert.assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
- Assert.assertEquals(streamConfig.getDecoderClass(), decoderClass);
- Assert.assertEquals(streamConfig.getDecoderProperties().size(), 0);
- Assert
- .assertEquals(streamConfig.getOffsetCriteria(), new OffsetCriteria.OffsetCriteriaBuilder().withOffsetLargest());
- Assert
- .assertEquals(streamConfig.getConnectionTimeoutMillis(), StreamConfig.DEFAULT_STREAM_CONNECTION_TIMEOUT_MILLIS);
- Assert.assertEquals(streamConfig.getFetchTimeoutMillis(), StreamConfig.DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
- Assert.assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
+ assertEquals(streamConfig.getType(), streamType);
+ assertEquals(streamConfig.getTopicName(), topic);
+ assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
+ assertEquals(streamConfig.getDecoderClass(), decoderClass);
+ assertEquals(streamConfig.getDecoderProperties().size(), 0);
+ assertEquals(streamConfig.getOffsetCriteria(), new OffsetCriteria.OffsetCriteriaBuilder().withOffsetLargest());
+ assertEquals(streamConfig.getConnectionTimeoutMillis(), StreamConfig.DEFAULT_STREAM_CONNECTION_TIMEOUT_MILLIS);
+ assertEquals(streamConfig.getFetchTimeoutMillis(), StreamConfig.DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS);
+ assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
+ assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES);
- consumerType = "lowLevel";
String offsetCriteria = "smallest";
String decoderProp1Key = "prop1";
String decoderProp1Value = "decoderValueString";
@@ -205,17 +157,13 @@ public class StreamConfigTest {
String flushThresholdTime = "2h";
String flushThresholdRows = "500";
String flushSegmentSize = "20M";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.DECODER_PROPS_PREFIX) + "."
+ decoderProp1Key, decoderProp1Value);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_OFFSET_CRITERIA), offsetCriteria);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS),
- connectionTimeout);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_OFFSET_CRITERIA), offsetCriteria);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS), connectionTimeout);
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_FETCH_TIMEOUT_MILLIS),
fetchTimeout);
@@ -224,42 +172,41 @@ public class StreamConfigTest {
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_SEGMENT_SIZE, flushSegmentSize);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getType(), streamType);
- Assert.assertEquals(streamConfig.getTopicName(), topic);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
- Assert.assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
- Assert.assertEquals(streamConfig.getDecoderClass(), decoderClass);
- Assert.assertEquals(streamConfig.getDecoderProperties().size(), 1);
- Assert.assertEquals(streamConfig.getDecoderProperties().get(decoderProp1Key), decoderProp1Value);
- Assert.assertTrue(streamConfig.getOffsetCriteria().isSmallest());
- Assert.assertEquals(streamConfig.getConnectionTimeoutMillis(), Long.parseLong(connectionTimeout));
- Assert.assertEquals(streamConfig.getFetchTimeoutMillis(), Integer.parseInt(fetchTimeout));
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(),
+ assertEquals(streamConfig.getType(), streamType);
+ assertEquals(streamConfig.getTopicName(), topic);
+ assertEquals(streamConfig.getConsumerFactoryClassName(), consumerFactoryClass);
+ assertEquals(streamConfig.getDecoderClass(), decoderClass);
+ assertEquals(streamConfig.getDecoderProperties().size(), 1);
+ assertEquals(streamConfig.getDecoderProperties().get(decoderProp1Key), decoderProp1Value);
+ assertTrue(streamConfig.getOffsetCriteria().isSmallest());
+ assertEquals(streamConfig.getConnectionTimeoutMillis(), Long.parseLong(connectionTimeout));
+ assertEquals(streamConfig.getFetchTimeoutMillis(), Integer.parseInt(fetchTimeout));
+ assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(),
(long) TimeUtils.convertPeriodToMillis(flushThresholdTime));
- Assert.assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(), DataSizeUtils.toBytes(flushSegmentSize));
+ assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(), DataSizeUtils.toBytes(flushSegmentSize));
// Backward compatibility check for flushThresholdTime
flushThresholdTime = "18000000";
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME, flushThresholdTime);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), Long.parseLong(flushThresholdTime));
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(), Long.parseLong(flushThresholdTime));
flushThresholdTime = "invalid input";
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME, flushThresholdTime);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
// Backward compatibility check for flush threshold rows
streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS);
streamConfigMap.put(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS, "10000");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), 10000);
+ assertEquals(streamConfig.getFlushThresholdRows(), 10000);
// Backward compatibility check for flush threshold segment size
streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_SEGMENT_SIZE);
streamConfigMap.put(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_DESIRED_SIZE, "10M");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(), DataSizeUtils.toBytes("10M"));
+ assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(), DataSizeUtils.toBytes("10M"));
}
/**
@@ -267,11 +214,8 @@ public class StreamConfigTest {
*/
@Test
public void testStreamConfigValidations() {
- boolean exception;
- StreamConfig streamConfig;
String streamType = "fakeStream";
String topic = "fakeTopic";
- String consumerType = "simple";
String tableName = "fakeTable_REALTIME";
String consumerFactoryClass = FakeStreamConsumerFactory.class.getName();
String decoderClass = FakeStreamMessageDecoder.class.getName();
@@ -279,82 +223,61 @@ public class StreamConfigTest {
// All mandatory properties set
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
- streamConfig = new StreamConfig(tableName, streamConfigMap);
-
- // Invalid consumer type
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- "invalidConsumerType");
- exception = false;
- try {
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- } catch (Exception e) {
- exception = true;
- }
- Assert.assertTrue(exception);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
+ new StreamConfig(tableName, streamConfigMap);
// Invalid fetch timeout
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_FETCH_TIMEOUT_MILLIS),
"timeout");
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFetchTimeoutMillis(), StreamConfig.DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS);
+ StreamConfig streamConfig = new StreamConfig(tableName, streamConfigMap);
+ assertEquals(streamConfig.getFetchTimeoutMillis(), StreamConfig.DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS);
// Invalid connection timeout
streamConfigMap.remove(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_FETCH_TIMEOUT_MILLIS));
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS), "timeout");
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS), "timeout");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert
- .assertEquals(streamConfig.getConnectionTimeoutMillis(), StreamConfig.DEFAULT_STREAM_CONNECTION_TIMEOUT_MILLIS);
+ assertEquals(streamConfig.getConnectionTimeoutMillis(), StreamConfig.DEFAULT_STREAM_CONNECTION_TIMEOUT_MILLIS);
// Invalid flush threshold rows - deprecated property
- streamConfigMap.remove(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS));
+ streamConfigMap.remove(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONNECTION_TIMEOUT_MILLIS));
streamConfigMap.put(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS, "rows");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
+ assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
// Invalid flush threshold rows - new property
streamConfigMap.remove(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS);
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS, "rows");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
+ assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
// Invalid flush threshold time
streamConfigMap.remove(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS);
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME, "time");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
// Invalid flush segment size - deprecated property
streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME);
streamConfigMap.put(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_DESIRED_SIZE, "size");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
+ assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES);
// Invalid flush segment size - new property
streamConfigMap.remove(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_DESIRED_SIZE);
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_SEGMENT_SIZE, "size");
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
+ assertEquals(streamConfig.getFlushThresholdSegmentSizeBytes(),
StreamConfig.DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES);
}
@@ -367,7 +290,6 @@ public class StreamConfigTest {
String streamType = "fakeStream";
String topic = "fakeTopic";
String tableName = "fakeTable_REALTIME";
- String consumerType = "lowlevel";
String consumerFactoryClass = FakeStreamConsumerFactory.class.getName();
String decoderClass = FakeStreamMessageDecoder.class.getName();
String flushThresholdRows = "200";
@@ -377,74 +299,45 @@ public class StreamConfigTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
-
- // use defaults if nothing provided
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
+
+ // Use default values if nothing provided
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
+ assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
- // use base values if provided
+ // Use regular values if provided
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS, flushThresholdRows);
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME, flushThresholdTime);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(),
+ assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(),
(long) TimeUtils.convertPeriodToMillis(flushThresholdTime));
- // llc overrides provided, but base values will be picked in base StreamConfigs
- streamConfigMap
- .put(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS + StreamConfigProperties.LLC_SUFFIX,
- flushThresholdRowsLLC);
+ // Use regular values if both regular and llc config exists
+ streamConfigMap.put(
+ StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS + StreamConfigProperties.LLC_SUFFIX,
+ flushThresholdRowsLLC);
streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME + StreamConfigProperties.LLC_SUFFIX,
flushThresholdTimeLLC);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(),
+ assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(),
(long) TimeUtils.convertPeriodToMillis(flushThresholdTime));
- // llc overrides provided, no base values, defaults will be picked
+ // Use llc values if only llc config exists
streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS);
streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME);
streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
- Assert.assertEquals(streamConfig.getFlushThresholdTimeMillis(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
-
- // PartitionLevel stream config will retrieve the llc overrides
- PartitionLevelStreamConfig partitionLevelStreamConfig = new PartitionLevelStreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRowsLLC));
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdTimeMillis(),
+ assertEquals(streamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRowsLLC));
+ assertEquals(streamConfig.getFlushThresholdTimeMillis(),
(long) TimeUtils.convertPeriodToMillis(flushThresholdTimeLLC));
-
- // PartitionLevelStreamConfig should use base values if llc overrides not provided
- streamConfigMap
- .remove(StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS + StreamConfigProperties.LLC_SUFFIX);
- streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME + StreamConfigProperties.LLC_SUFFIX);
- streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS, flushThresholdRows);
- streamConfigMap.put(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME, flushThresholdTime);
- partitionLevelStreamConfig = new PartitionLevelStreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdRows(), Integer.parseInt(flushThresholdRows));
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdTimeMillis(),
- (long) TimeUtils.convertPeriodToMillis(flushThresholdTime));
-
- // PartitionLevelStreamConfig should use defaults if nothing provided
- streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_ROWS);
- streamConfigMap.remove(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME);
- partitionLevelStreamConfig = new PartitionLevelStreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdRows(), StreamConfig.DEFAULT_FLUSH_THRESHOLD_ROWS);
- Assert.assertEquals(partitionLevelStreamConfig.getFlushThresholdTimeMillis(),
- StreamConfig.DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
}
@Test
@@ -457,63 +350,55 @@ public class StreamConfigTest {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
- topic);
- streamConfigMap.put(StreamConfigProperties
- .constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
- consumerFactoryClass);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
- decoderClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
+ streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
+ StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClass);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_DECODER_CLASS),
+ decoderClass);
String consumerType = "simple";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- StreamConfig streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
- Assert.assertTrue(streamConfig.hasLowLevelConsumerType());
- Assert.assertFalse(streamConfig.hasHighLevelConsumerType());
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
+ consumerType);
+ new StreamConfig(tableName, streamConfigMap);
consumerType = "lowlevel";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
- streamConfig = new StreamConfig(tableName, streamConfigMap);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
- Assert.assertTrue(streamConfig.hasLowLevelConsumerType());
- Assert.assertFalse(streamConfig.hasHighLevelConsumerType());
-
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
+ consumerType);
+ new StreamConfig(tableName, streamConfigMap);
try {
consumerType = "highLevel";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
+ consumerType);
new StreamConfig(tableName, streamConfigMap);
- Assert.fail("Invalid consumer type(s) " + consumerType + " in stream config");
+ fail("Invalid consumer type(s) " + consumerType + " in stream config");
} catch (Exception e) {
// expected
}
try {
consumerType = "highLevel,simple";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
+ consumerType);
new StreamConfig(tableName, streamConfigMap);
- Assert.fail("Invalid consumer type(s) " + consumerType + " in stream config");
+ fail("Invalid consumer type(s) " + consumerType + " in stream config");
} catch (Exception e) {
// expected
}
try {
consumerType = "highLevel,lowlevel";
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
+ streamConfigMap.put(
+ StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
+ consumerType);
new StreamConfig(tableName, streamConfigMap);
+ fail("Invalid consumer type(s) " + consumerType + " in stream config");
} catch (Exception e) {
// expected
}
diff --git a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
index 6b6cc9202c..41ece0290b 100644
--- a/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
+++ b/pinot-integration-test-base/src/test/java/org/apache/pinot/integration/tests/BaseClusterIntegrationTest.java
@@ -58,7 +58,6 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamDataProducer;
import org.apache.pinot.spi.stream.StreamDataProvider;
@@ -86,12 +85,9 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
"On_Time_On_Time_Performance_2014_100k_subset_nonulls.tar.gz";
protected static final long DEFAULT_COUNT_STAR_RESULT = 115545L;
protected static final int DEFAULT_LLC_SEGMENT_FLUSH_SIZE = 5000;
- protected static final int DEFAULT_HLC_SEGMENT_FLUSH_SIZE = 20000;
protected static final int DEFAULT_TRANSACTION_NUM_KAFKA_BROKERS = 3;
protected static final int DEFAULT_LLC_NUM_KAFKA_BROKERS = 2;
- protected static final int DEFAULT_HLC_NUM_KAFKA_BROKERS = 1;
protected static final int DEFAULT_LLC_NUM_KAFKA_PARTITIONS = 2;
- protected static final int DEFAULT_HLC_NUM_KAFKA_PARTITIONS = 10;
protected static final int DEFAULT_MAX_NUM_KAFKA_MESSAGES_PER_BATCH = 10000;
protected static final List<String> DEFAULT_NO_DICTIONARY_COLUMNS =
Arrays.asList("ActualElapsedTime", "ArrDelay", "DepDelay", "CRSDepTime");
@@ -141,10 +137,6 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
return DEFAULT_COUNT_STAR_RESULT;
}
- protected boolean useLlc() {
- return true;
- }
-
protected boolean useKafkaTransaction() {
return false;
}
@@ -154,22 +146,11 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
}
protected int getRealtimeSegmentFlushSize() {
- if (useLlc()) {
- return DEFAULT_LLC_SEGMENT_FLUSH_SIZE;
- } else {
- return DEFAULT_HLC_SEGMENT_FLUSH_SIZE;
- }
+ return DEFAULT_LLC_SEGMENT_FLUSH_SIZE;
}
protected int getNumKafkaBrokers() {
- if (useKafkaTransaction()) {
- return DEFAULT_TRANSACTION_NUM_KAFKA_BROKERS;
- }
- if (useLlc()) {
- return DEFAULT_LLC_NUM_KAFKA_BROKERS;
- } else {
- return DEFAULT_HLC_NUM_KAFKA_BROKERS;
- }
+ return useKafkaTransaction() ? DEFAULT_TRANSACTION_NUM_KAFKA_BROKERS : DEFAULT_LLC_NUM_KAFKA_BROKERS;
}
protected int getKafkaPort() {
@@ -182,11 +163,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
}
protected int getNumKafkaPartitions() {
- if (useLlc()) {
- return DEFAULT_LLC_NUM_KAFKA_PARTITIONS;
- } else {
- return DEFAULT_HLC_NUM_KAFKA_PARTITIONS;
- }
+ return DEFAULT_LLC_NUM_KAFKA_PARTITIONS;
}
protected String getKafkaTopic() {
@@ -347,30 +324,13 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
Map<String, String> streamConfigMap = new HashMap<>();
String streamType = "kafka";
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
- boolean useLlc = useLlc();
- if (useLlc) {
- // LLC
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.LOWLEVEL.toString());
- streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
- KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BROKER_LIST),
- "localhost:" + _kafkaStarters.get(0).getPort());
- if (useKafkaTransaction()) {
- streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
- KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL),
- KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL_READ_COMMITTED);
- }
- } else {
- // HLC
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.HIGHLEVEL.toString());
- streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
- KafkaStreamConfigProperties.HighLevelConsumer.KAFKA_HLC_ZK_CONNECTION_STRING), getKafkaZKAddress());
+ streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
+ KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_BROKER_LIST),
+ "localhost:" + _kafkaStarters.get(0).getPort());
+ if (useKafkaTransaction()) {
streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(
- KafkaStreamConfigProperties.HighLevelConsumer.KAFKA_HLC_BOOTSTRAP_SERVER),
- "localhost:" + _kafkaStarters.get(0).getPort());
+ KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL),
+ KafkaStreamConfigProperties.LowLevelConsumer.KAFKA_ISOLATION_LEVEL_READ_COMMITTED);
}
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), getStreamConsumerFactoryClassName());
@@ -399,7 +359,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
.setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion())
.setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig()).setBrokerTenant(getBrokerTenant())
.setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()).setQueryConfig(getQueryConfig())
- .setLLC(useLlc()).setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
+ .setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
}
/**
@@ -418,7 +378,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
.setTimeColumnName(getTimeColumnName()).setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas())
.setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig())
.setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig())
- .setLLC(useLlc()).setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled())
+ .setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled())
.setRoutingConfig(new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE))
.setSegmentPartitionConfig(new SegmentPartitionConfig(columnPartitionConfigMap))
.setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
@@ -472,7 +432,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
.setTimeColumnName(getTimeColumnName()).setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas())
.setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig())
.setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig())
- .setLLC(useLlc()).setStreamConfigs(streamConfigsMap)
+ .setStreamConfigs(streamConfigsMap)
.setNullHandlingEnabled(UpsertConfig.Mode.PARTIAL.equals(upsertConfig.getMode()) || getNullHandlingEnabled())
.setRoutingConfig(new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE))
.setSegmentPartitionConfig(new SegmentPartitionConfig(columnPartitionConfigMap))
@@ -492,7 +452,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
.setTimeColumnName(getTimeColumnName()).setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas())
.setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig())
.setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig())
- .setLLC(useLlc()).setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled())
+ .setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled())
.setRoutingConfig(new RoutingConfig(null, null, RoutingConfig.STRICT_REPLICA_GROUP_INSTANCE_SELECTOR_TYPE))
.setSegmentPartitionConfig(new SegmentPartitionConfig(columnPartitionConfigMap))
.setReplicaGroupStrategyConfig(new ReplicaGroupStrategyConfig(primaryKeyColumn, 1))
@@ -801,11 +761,7 @@ public abstract class BaseClusterIntegrationTest extends ClusterTest {
}
protected String getType(JsonNode jsonNode, int colIndex) {
- return jsonNode.get("resultTable")
- .get("dataSchema")
- .get("columnDataTypes")
- .get(colIndex)
- .asText();
+ return jsonNode.get("resultTable").get("dataSchema").get("columnDataTypes").get(colIndex).asText();
}
protected <T> T getCellValue(JsonNode jsonNode, int colIndex, int rowIndex, Function<JsonNode, T> extract) {
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
index 6874da8362..2d5a8109c2 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
@@ -243,9 +243,9 @@ public class MergeRollupMinionClusterIntegrationTest extends BaseClusterIntegrat
.setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion())
.setLoadMode(getLoadMode()).setTaskConfig(
new TableTaskConfig(Collections.singletonMap(MinionConstants.MergeRollupTask.TASK_TYPE, tableTaskConfigs)))
- .setBrokerTenant(getBrokerTenant())
- .setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()).setQueryConfig(getQueryConfig())
- .setLLC(useLlc()).setStreamConfigs(streamConfigs).setNullHandlingEnabled(getNullHandlingEnabled()).build();
+ .setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig())
+ .setQueryConfig(getQueryConfig()).setStreamConfigs(streamConfigs)
+ .setNullHandlingEnabled(getNullHandlingEnabled()).build();
}
private TableTaskConfig getSingleLevelConcatTaskConfig() {
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeKinesisIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeKinesisIntegrationTest.java
index 8bb4767fb3..f2f6602766 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeKinesisIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeKinesisIntegrationTest.java
@@ -54,7 +54,6 @@ import org.apache.pinot.plugin.stream.kinesis.KinesisConsumerFactory;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.StringUtil;
@@ -179,8 +178,7 @@ public class RealtimeKinesisIntegrationTest extends BaseClusterIntegrationTestSe
.setTimeColumnName("DaysSinceEpoch").setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas())
.setSegmentVersion(getSegmentVersion()).setLoadMode(getLoadMode()).setTaskConfig(getTaskConfig())
.setBrokerTenant(getBrokerTenant()).setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig())
- .setLLC(true).setStreamConfigs(createKinesisStreamConfig()).setNullHandlingEnabled(getNullHandlingEnabled())
- .build();
+ .setStreamConfigs(createKinesisStreamConfig()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
}
public Map<String, String> createKinesisStreamConfig() {
@@ -195,9 +193,6 @@ public class RealtimeKinesisIntegrationTest extends BaseClusterIntegrationTestSe
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_FETCH_TIMEOUT_MILLIS),
"30000");
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.LOWLEVEL.toString());
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE,
StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), KinesisConsumerFactory.class.getName());
streamConfigMap.put(
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
index 82aa58659b..ddfac35562 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/RealtimeToOfflineSegmentsMinionClusterIntegrationTest.java
@@ -210,7 +210,7 @@ public class RealtimeToOfflineSegmentsMinionClusterIntegrationTest extends BaseC
.setFieldConfigList(getFieldConfigs()).setNumReplicas(getNumReplicas()).setSegmentVersion(getSegmentVersion())
.setLoadMode(getLoadMode()).setTaskConfig(taskConfig).setBrokerTenant(getBrokerTenant())
.setServerTenant(getServerTenant()).setIngestionConfig(getIngestionConfig()).setQueryConfig(getQueryConfig())
- .setLLC(useLlc()).setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
+ .setStreamConfigs(getStreamConfigs()).setNullHandlingEnabled(getNullHandlingEnabled()).build();
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentPreloadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentPreloadIntegrationTest.java
index 47cebf3c24..1a83754b96 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentPreloadIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentPreloadIntegrationTest.java
@@ -217,7 +217,7 @@ public class UpsertTableSegmentPreloadIntegrationTest extends BaseClusterIntegra
for (BaseServerStarter serverStarter : _serverStarters) {
String segmentDir = serverStarter.getConfig().getProperty(CommonConstants.Server.CONFIG_OF_INSTANCE_DATA_DIR);
File[] files = new File(segmentDir, realtimeTableName).listFiles(
- (dir, name) -> name.startsWith(rawTableName) && !LLCSegmentName.isLowLevelConsumerSegmentName(name));
+ (dir, name) -> name.startsWith(rawTableName) && !LLCSegmentName.isLLCSegment(name));
for (File file : files) {
if (!new File(new File(file, "v3"), V1Constants.VALID_DOC_IDS_SNAPSHOT_FILE_NAME).exists()) {
return false;
@@ -239,8 +239,8 @@ public class UpsertTableSegmentPreloadIntegrationTest extends BaseClusterIntegra
int maxSequenceNumber = 0;
for (Map.Entry<String, Map<String, String>> entry : segmentAssignment.entrySet()) {
String segmentName = entry.getKey();
- if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
maxSequenceNumber = Math.max(maxSequenceNumber, llcSegmentName.getSequenceNumber());
}
}
@@ -253,8 +253,8 @@ public class UpsertTableSegmentPreloadIntegrationTest extends BaseClusterIntegra
assertEquals(instanceStateMap.size(), 1);
Map.Entry<String, String> instanceIdAndState = instanceStateMap.entrySet().iterator().next();
String state = instanceIdAndState.getValue();
- if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
+ if (llcSegmentName != null) {
if (llcSegmentName.getSequenceNumber() < maxSequenceNumber) {
assertEquals(state, SegmentStateModel.ONLINE);
} else {
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
index 2b9f94ae10..28edcb23b3 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/UpsertTableSegmentUploadIntegrationTest.java
@@ -196,7 +196,7 @@ public class UpsertTableSegmentUploadIntegrationTest extends BaseClusterIntegrat
assertEquals(instanceStateMap.size(), 1);
Map.Entry<String, String> instanceIdAndState = instanceStateMap.entrySet().iterator().next();
String state = instanceIdAndState.getValue();
- if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ if (LLCSegmentName.isLLCSegment(segmentName)) {
assertEquals(state, SegmentStateModel.CONSUMING);
} else {
assertEquals(state, SegmentStateModel.ONLINE);
diff --git a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java
index ff65c704ba..51be9b1ce3 100644
--- a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java
+++ b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGenerator.java
@@ -42,9 +42,7 @@ import org.apache.pinot.spi.annotations.minion.TaskGenerator;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.CommonConstants.Segment;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -101,12 +99,6 @@ public class RealtimeToOfflineSegmentsTaskGenerator extends BaseTaskGenerator {
LOGGER.warn("Skip generating task: {} for non-REALTIME table: {}", taskType, realtimeTableName);
continue;
}
- StreamConfig streamConfig =
- new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(tableConfig));
- if (streamConfig.hasHighLevelConsumerType()) {
- LOGGER.warn("Skip generating task: {} for HLC REALTIME table: {}", taskType, realtimeTableName);
- continue;
- }
LOGGER.info("Start generating task configs for table: {} for task: {}", realtimeTableName, taskType);
// Only schedule 1 task of this type, per table
diff --git a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGeneratorTest.java b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGeneratorTest.java
index 000383b7ff..0fe1fb8c89 100644
--- a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGeneratorTest.java
+++ b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/test/java/org/apache/pinot/plugin/minion/tasks/realtimetoofflinesegments/RealtimeToOfflineSegmentsTaskGeneratorTest.java
@@ -34,7 +34,6 @@ import org.apache.pinot.core.minion.PinotTaskConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableTaskConfig;
import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.utils.CommonConstants.Segment.Realtime.Status;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
@@ -61,9 +60,6 @@ public class RealtimeToOfflineSegmentsTaskGeneratorTest {
@BeforeClass
public void setup() {
_streamConfigs.put(StreamConfigProperties.STREAM_TYPE, "kafka");
- _streamConfigs
- .put(StreamConfigProperties.constructStreamProperty("kafka", StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.LOWLEVEL.toString());
_streamConfigs
.put(StreamConfigProperties.constructStreamProperty("kafka", StreamConfigProperties.STREAM_TOPIC_NAME),
"myTopic");
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java
index 4159595f39..a781cb408b 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelStreamConfigTest.java
@@ -52,15 +52,11 @@ public class KafkaPartitionLevelStreamConfigTest {
String populateRowMetadata) {
Map<String, String> streamConfigMap = new HashMap<>();
String streamType = "kafka";
- String consumerType = StreamConfig.ConsumerType.LOWLEVEL.toString();
String consumerFactoryClassName = KafkaConsumerFactory.class.getName();
String tableNameWithType = "tableName_REALTIME";
streamConfigMap.put(StreamConfigProperties.STREAM_TYPE, streamType);
streamConfigMap.put(
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME), topic);
- streamConfigMap.put(
- StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
streamConfigMap.put(StreamConfigProperties.constructStreamProperty(streamType,
StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS), consumerFactoryClassName);
streamConfigMap.put(
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumerTest.java
index b09820530e..5d620de604 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisConsumerTest.java
@@ -68,8 +68,6 @@ public class KinesisConsumerTest {
props.put(StreamConfigProperties.STREAM_TYPE, STREAM_TYPE);
props.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_TOPIC_NAME),
STREAM_NAME);
- props.put(StreamConfigProperties.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- StreamConfig.ConsumerType.LOWLEVEL.toString());
props.put(StreamConfigProperties
.constructStreamProperty(STREAM_TYPE, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
KinesisConsumerFactory.class.getName());
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/readerwriter/RealtimeIndexOffHeapMemoryManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/readerwriter/RealtimeIndexOffHeapMemoryManager.java
index 737460593c..88f756bab1 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/readerwriter/RealtimeIndexOffHeapMemoryManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/io/readerwriter/RealtimeIndexOffHeapMemoryManager.java
@@ -25,9 +25,7 @@ import java.util.List;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.pinot.common.metrics.ServerGauge;
import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.common.utils.HLCSegmentName;
import org.apache.pinot.common.utils.LLCSegmentName;
-import org.apache.pinot.common.utils.SegmentName;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
@@ -58,9 +56,6 @@ public abstract class RealtimeIndexOffHeapMemoryManager implements PinotDataBuff
LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
if (llcSegmentName != null) {
_rawTableName = llcSegmentName.getTableName();
- } else if (SegmentName.isHighLevelConsumerSegmentName(segmentName)) {
- HLCSegmentName hlcSegmentName = new HLCSegmentName(segmentName);
- _rawTableName = hlcSegmentName.getTableName();
} else {
// For testing only
_rawTableName = "NoSuchTable";
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
index 089ad01430..35b6aed5fb 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java
@@ -133,8 +133,8 @@ public class ConcurrentMapPartitionUpsertMetadataManager extends BasePartitionUp
// Update the record location when getting a newer comparison value, or the value is the same as the
// current value, but the segment has a larger sequence number (the segment is newer than the current
// segment).
- if (comparisonResult > 0 || (comparisonResult == 0 && LLCSegmentName.isLowLevelConsumerSegmentName(
- segmentName) && LLCSegmentName.isLowLevelConsumerSegmentName(currentSegmentName)
+ if (comparisonResult > 0 || (comparisonResult == 0 && LLCSegmentName.isLLCSegment(segmentName)
+ && LLCSegmentName.isLLCSegment(currentSegmentName)
&& LLCSegmentName.getSequenceNumber(segmentName) > LLCSegmentName.getSequenceNumber(
currentSegmentName))) {
removeDocId(currentSegment, currentDocId);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReplicationUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReplicationUtils.java
deleted file mode 100644
index 3a601e1ca6..0000000000
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ReplicationUtils.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * 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.segment.local.utils;
-
-import org.apache.pinot.spi.config.table.SegmentsValidationAndRetentionConfig;
-import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.config.table.TableType;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
-
-
-/**
- * Methods related to replication
- */
-public class ReplicationUtils {
- private ReplicationUtils() {
- }
-
- /**
- * Decides if {@link SegmentsValidationAndRetentionConfig ::getReplicationNumber} should be used
- */
- public static boolean useReplication(TableConfig tableConfig) {
-
- TableType tableType = tableConfig.getTableType();
- if (tableType.equals(TableType.REALTIME)) {
- StreamConfig streamConfig =
- new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
- return streamConfig.hasHighLevelConsumerType();
- }
- return true;
- }
-
- /**
- * Decides if {@link SegmentsValidationAndRetentionConfig ::getReplicasPerPartitionNumber} should be used
- */
- public static boolean useReplicasPerPartition(TableConfig tableConfig) {
-
- TableType tableType = tableConfig.getTableType();
- if (tableType.equals(TableType.REALTIME)) {
- StreamConfig streamConfig =
- new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
- return streamConfig.hasLowLevelConsumerType();
- }
- return false;
- }
-}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
index 7d1b5d6ca3..413a2d7eec 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java
@@ -1164,53 +1164,16 @@ public final class TableConfigUtils {
}
/**
- * TODO: After deprecating "replicasPerPartition", we can change this function's behavior to always overwrite
- * config to "replication" only.
- *
* Ensure that the table config has the minimum number of replicas set as per cluster configs.
- * If is doesn't, set the required amount of replication in the table config
*/
public static void ensureMinReplicas(TableConfig tableConfig, int defaultTableMinReplicas) {
- // For self-serviced cluster, ensure that the tables are created with at least min replication factor irrespective
- // of table configuration value
- SegmentsValidationAndRetentionConfig segmentsConfig = tableConfig.getValidationConfig();
- boolean verifyReplicasPerPartition;
- boolean verifyReplication;
-
- try {
- verifyReplicasPerPartition = ReplicationUtils.useReplicasPerPartition(tableConfig);
- verifyReplication = ReplicationUtils.useReplication(tableConfig);
- } catch (Exception e) {
- throw new IllegalStateException(String.format("Invalid tableIndexConfig or streamConfig: %s", e.getMessage()), e);
- }
-
- if (verifyReplication) {
- int requestReplication;
- try {
- requestReplication = tableConfig.getReplication();
- if (requestReplication < defaultTableMinReplicas) {
- LOGGER.info("Creating table with minimum replication factor of: {} instead of requested replication: {}",
- defaultTableMinReplicas, requestReplication);
- segmentsConfig.setReplication(String.valueOf(defaultTableMinReplicas));
- }
- } catch (NumberFormatException e) {
- throw new IllegalStateException("Invalid replication number", e);
- }
- }
-
- if (verifyReplicasPerPartition) {
- int replicasPerPartition;
- try {
- replicasPerPartition = tableConfig.getReplication();
- if (replicasPerPartition < defaultTableMinReplicas) {
- LOGGER.info(
- "Creating table with minimum replicasPerPartition of: {} instead of requested replicasPerPartition: {}",
- defaultTableMinReplicas, replicasPerPartition);
- segmentsConfig.setReplicasPerPartition(String.valueOf(defaultTableMinReplicas));
- }
- } catch (NumberFormatException e) {
- throw new IllegalStateException("Invalid replicasPerPartition number", e);
- }
+ SegmentsValidationAndRetentionConfig validationConfig = tableConfig.getValidationConfig();
+ int replication = tableConfig.getReplication();
+ if (replication < defaultTableMinReplicas) {
+ LOGGER.info("Creating table with minimum replication factor of: {} instead of requested replication: {}",
+ defaultTableMinReplicas, replication);
+ validationConfig.setReplicasPerPartition(null);
+ validationConfig.setReplication(String.valueOf(defaultTableMinReplicas));
}
}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
index c03328c2bb..0e261e143d 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
@@ -565,7 +565,7 @@ public class TablesResource {
}
// Check the segment is low level consumer segment
- if (!LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
+ if (!LLCSegmentName.isLLCSegment(segmentName)) {
throw new WebApplicationException(String.format("Segment %s is not a low level consumer segment", segmentName),
Response.Status.BAD_REQUEST);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
index 849c1e8902..463c300f3c 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/SegmentsValidationAndRetentionConfig.java
@@ -35,7 +35,9 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
@Deprecated
private String _segmentPushType;
private String _replication;
- // For high-level consumers, the number of replicas should be same as num server instances
+ @Deprecated // Use _replication instead
+ private String _replicasPerPartition;
+ @Deprecated // Schema name should be the same as raw table name
private String _schemaName;
private String _timeColumnName;
private TimeUnit _timeType;
@@ -49,9 +51,6 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
// For more usage of this field, please refer to this design doc: https://tinyurl.com/f63ru4sb
private String _peerSegmentDownloadScheme;
- // Number of replicas per partition of low-level consumers. This config is used for realtime tables only.
- private String _replicasPerPartition;
-
public String getSegmentAssignmentStrategy() {
return _segmentAssignmentStrategy;
}
@@ -105,10 +104,12 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
/**
* @deprecated Use {@code segmentIngestionFrequency} from {@link IngestionConfig#getBatchIngestionConfig()}
*/
+ @Deprecated
public String getSegmentPushFrequency() {
return _segmentPushFrequency;
}
+ @Deprecated
public void setSegmentPushFrequency(String segmentPushFrequency) {
_segmentPushFrequency = segmentPushFrequency;
}
@@ -116,10 +117,12 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
/**
* @deprecated Use {@code segmentIngestionType} from {@link IngestionConfig#getBatchIngestionConfig()}
*/
+ @Deprecated
public String getSegmentPushType() {
return _segmentPushType;
}
+ @Deprecated
public void setSegmentPushType(String segmentPushType) {
_segmentPushType = segmentPushType;
}
@@ -135,27 +138,33 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
_replication = replication;
}
- // Schema name should be the same as raw table name
- @Deprecated
- public String getSchemaName() {
- return _schemaName;
- }
-
- public void setSchemaName(String schemaName) {
- _schemaName = schemaName;
- }
-
/**
* Try to Use {@link TableConfig#getReplication()}
+ * @deprecated Use _replication instead
*/
+ @Deprecated
public String getReplicasPerPartition() {
return _replicasPerPartition;
}
+ @Deprecated
public void setReplicasPerPartition(String replicasPerPartition) {
_replicasPerPartition = replicasPerPartition;
}
+ /**
+ * @deprecated Schema name should be the same as raw table name
+ */
+ @Deprecated
+ public String getSchemaName() {
+ return _schemaName;
+ }
+
+ @Deprecated
+ public void setSchemaName(String schemaName) {
+ _schemaName = schemaName;
+ }
+
public ReplicaGroupStrategyConfig getReplicaGroupStrategyConfig() {
return _replicaGroupStrategyConfig;
}
@@ -175,6 +184,7 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
/**
* Try to Use {@link TableConfig#getReplication()}
*/
+ @Deprecated
@JsonIgnore
public int getReplicationNumber() {
return Integer.parseInt(_replication);
@@ -183,6 +193,7 @@ public class SegmentsValidationAndRetentionConfig extends BaseJsonConfig {
/**
* Try to Use {@link TableConfig#getReplication()}
*/
+ @Deprecated
@JsonIgnore
public int getReplicasPerPartitionNumber() {
return Integer.parseInt(_replicasPerPartition);
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
index f16d2e67f9..c71b5481d5 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java
@@ -31,8 +31,6 @@ import org.apache.pinot.spi.config.table.assignment.InstanceAssignmentConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.table.assignment.SegmentAssignmentConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
-import org.apache.pinot.spi.stream.StreamConfig;
-import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
@@ -415,27 +413,13 @@ public class TableConfig extends BaseJsonConfig {
@JsonIgnore
public int getReplication() {
- int replication = 0;
if (_tableType == TableType.REALTIME) {
- StreamConfig streamConfig = new StreamConfig(_tableName, IngestionConfigUtils.getStreamConfigMap(this));
- if (streamConfig.hasHighLevelConsumerType()) {
- // In case of HLC, we read from "replication"
- replication = Integer.parseInt(_validationConfig.getReplication());
- } else {
- // To keep the backward compatibility, we read from "replicasPerPartition" in case of LLC
- String replicasPerPartitionStr = _validationConfig.getReplicasPerPartition();
- try {
- replication = Integer.parseInt(replicasPerPartitionStr);
- } catch (NumberFormatException e) {
- // If numReplicasPerPartition is not being used or specified, read the value from replication
- String replicationStr = _validationConfig.getReplication();
- replication = Integer.parseInt(replicationStr);
- }
+ // Use replicasPerPartition for real-time table if exists
+ String replicasPerPartition = _validationConfig.getReplicasPerPartition();
+ if (replicasPerPartition != null) {
+ return Integer.parseInt(replicasPerPartition);
}
- } else {
- // In case of OFFLINE tables, we read from "replication"
- replication = Integer.parseInt(_validationConfig.getReplication());
}
- return replication;
+ return Integer.parseInt(_validationConfig.getReplication());
}
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
index e545b94617..82092fe336 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionLevelStreamConfig.java
@@ -18,62 +18,13 @@
*/
package org.apache.pinot.spi.stream;
-import com.google.common.base.Preconditions;
import java.util.Map;
-import org.apache.pinot.spi.utils.TimeUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-/**
- * A {@link StreamConfig} for LLC partition level stream, which overrides some properties for low-level consumer.
- * This can be removed once we remove HLC implementation from the code
- */
+@Deprecated
public class PartitionLevelStreamConfig extends StreamConfig {
- private static final Logger LOGGER = LoggerFactory.getLogger(PartitionLevelStreamConfig.class);
public PartitionLevelStreamConfig(String tableNameWithType, Map<String, String> streamConfigMap) {
super(tableNameWithType, streamConfigMap);
}
-
- @Override
- protected int extractFlushThresholdRows(Map<String, String> streamConfigMap) {
- String flushThresholdRowsKey =
- StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS + StreamConfigProperties.LLC_SUFFIX;
- String flushThresholdRowsStr = streamConfigMap.get(flushThresholdRowsKey);
- if (flushThresholdRowsStr != null) {
- try {
- int flushThresholdRows = Integer.parseInt(flushThresholdRowsStr);
- // Flush threshold rows 0 means using segment size based flush threshold
- Preconditions.checkState(flushThresholdRows >= 0);
- return flushThresholdRows;
- } catch (Exception e) {
- int defaultValue = super.extractFlushThresholdRows(streamConfigMap);
- LOGGER.warn("Invalid config {}: {}, defaulting to: {}", flushThresholdRowsKey, flushThresholdRowsStr,
- defaultValue);
- return defaultValue;
- }
- } else {
- return super.extractFlushThresholdRows(streamConfigMap);
- }
- }
-
- @Override
- protected long extractFlushThresholdTimeMillis(Map<String, String> streamConfigMap) {
- String flushThresholdTimeKey =
- StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME + StreamConfigProperties.LLC_SUFFIX;
- String flushThresholdTimeStr = streamConfigMap.get(flushThresholdTimeKey);
- if (flushThresholdTimeStr != null) {
- try {
- return TimeUtils.convertPeriodToMillis(flushThresholdTimeStr);
- } catch (Exception e) {
- long defaultValue = super.extractFlushThresholdTimeMillis(streamConfigMap);
- LOGGER.warn("Invalid config {}: {}, defaulting to: {}", flushThresholdTimeKey, flushThresholdTimeStr,
- defaultValue);
- return defaultValue;
- }
- } else {
- return super.extractFlushThresholdTimeMillis(streamConfigMap);
- }
- }
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java
index 54b0c7905b..94f8adf566 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfig.java
@@ -19,14 +19,13 @@
package org.apache.pinot.spi.stream;
import com.google.common.base.Preconditions;
-import java.util.ArrayList;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.utils.DataSizeUtils;
-import org.apache.pinot.spi.utils.EqualityUtils;
import org.apache.pinot.spi.utils.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -38,14 +37,6 @@ import org.slf4j.LoggerFactory;
public class StreamConfig {
private static final Logger LOGGER = LoggerFactory.getLogger(StreamConfig.class);
- /**
- * The type of the stream consumer either HIGHLEVEL or LOWLEVEL. For backward compatibility, adding SIMPLE which is
- * equivalent to LOWLEVEL
- */
- public enum ConsumerType {
- HIGHLEVEL, LOWLEVEL
- }
-
public static final int DEFAULT_FLUSH_THRESHOLD_ROWS = 5_000_000;
public static final long DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS = TimeUnit.MILLISECONDS.convert(6, TimeUnit.HOURS);
public static final long DEFAULT_FLUSH_THRESHOLD_SEGMENT_SIZE_BYTES = 200 * 1024 * 1024; // 200M
@@ -59,14 +50,11 @@ public class StreamConfig {
public static final int DEFAULT_STREAM_FETCH_TIMEOUT_MILLIS = 5_000;
public static final int DEFAULT_IDLE_TIMEOUT_MILLIS = 3 * 60 * 1000;
- private static final String SIMPLE_CONSUMER_TYPE_STRING = "simple";
-
private static final double CONSUMPTION_RATE_LIMIT_NOT_SPECIFIED = -1;
private final String _type;
private final String _topicName;
private final String _tableNameWithType;
- private final List<ConsumerType> _consumerTypes = new ArrayList<>();
private final String _consumerFactoryClassName;
private final String _decoderClass;
private final Map<String, String> _decoderProperties = new HashMap<>();
@@ -110,7 +98,6 @@ public class StreamConfig {
_tableNameWithType = tableNameWithType;
validateConsumerType(_type, streamConfigMap);
- _consumerTypes.add(ConsumerType.LOWLEVEL);
String consumerFactoryClassKey =
StreamConfigProperties.constructStreamProperty(_type, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS);
@@ -136,8 +123,8 @@ public class StreamConfig {
StreamConfigProperties.constructStreamProperty(_type, StreamConfigProperties.DECODER_PROPS_PREFIX);
for (String key : streamConfigMap.keySet()) {
if (key.startsWith(streamDecoderPropPrefix)) {
- _decoderProperties
- .put(StreamConfigProperties.getPropertySuffix(key, streamDecoderPropPrefix), streamConfigMap.get(key));
+ _decoderProperties.put(StreamConfigProperties.getPropertySuffix(key, streamDecoderPropPrefix),
+ streamConfigMap.get(key));
}
}
@@ -188,7 +175,7 @@ public class StreamConfig {
_flushThresholdSegmentSizeBytes = extractFlushThresholdSegmentSize(streamConfigMap);
_serverUploadToDeepStore = Boolean.parseBoolean(
streamConfigMap.getOrDefault(StreamConfigProperties.SERVER_UPLOAD_TO_DEEPSTORE,
- DEFAULT_SERVER_UPLOAD_TO_DEEPSTORE));
+ DEFAULT_SERVER_UPLOAD_TO_DEEPSTORE));
int autotuneInitialRows = 0;
String initialRowsValue = streamConfigMap.get(StreamConfigProperties.SEGMENT_FLUSH_AUTOTUNE_INITIAL_ROWS);
@@ -216,13 +203,15 @@ public class StreamConfig {
String consumerTypesKey =
StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES);
String consumerTypes = streamConfigMap.get(consumerTypesKey);
- Preconditions.checkNotNull(consumerTypes, consumerTypesKey + " cannot be null");
- for (String consumerType : consumerTypes.split(",")) {
- Preconditions.checkState(ConsumerType.LOWLEVEL.name().equalsIgnoreCase(consumerType)
- || SIMPLE_CONSUMER_TYPE_STRING.equalsIgnoreCase(consumerType),
+ if (consumerTypes == null) {
+ return;
+ }
+ for (String consumerType : StringUtils.split(consumerTypes, ',')) {
+ Preconditions.checkState(!consumerType.equalsIgnoreCase("highlevel"),
"Realtime tables with HLC consumer (consumer.type=highlevel) is no longer supported in Apache Pinot");
}
}
+
public boolean isServerUploadToDeepStore() {
return _serverUploadToDeepStore;
}
@@ -260,6 +249,11 @@ public class StreamConfig {
key = StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS;
flushThresholdRowsStr = streamConfigMap.get(key);
}
+ if (flushThresholdRowsStr == null) {
+ // for backward compatibility with older property
+ key = StreamConfigProperties.DEPRECATED_SEGMENT_FLUSH_THRESHOLD_ROWS + StreamConfigProperties.LLC_SUFFIX;
+ flushThresholdRowsStr = streamConfigMap.get(key);
+ }
if (flushThresholdRowsStr != null) {
try {
int flushThresholdRows = Integer.parseInt(flushThresholdRowsStr);
@@ -267,8 +261,8 @@ public class StreamConfig {
Preconditions.checkState(flushThresholdRows >= 0);
return flushThresholdRows;
} catch (Exception e) {
- LOGGER
- .warn("Invalid config {}: {}, defaulting to: {}", key, flushThresholdRowsStr, DEFAULT_FLUSH_THRESHOLD_ROWS);
+ LOGGER.warn("Invalid config {}: {}, defaulting to: {}", key, flushThresholdRowsStr,
+ DEFAULT_FLUSH_THRESHOLD_ROWS);
return DEFAULT_FLUSH_THRESHOLD_ROWS;
}
} else {
@@ -277,7 +271,13 @@ public class StreamConfig {
}
protected long extractFlushThresholdTimeMillis(Map<String, String> streamConfigMap) {
- String flushThresholdTimeStr = streamConfigMap.get(StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME);
+ String key = StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME;
+ String flushThresholdTimeStr = streamConfigMap.get(key);
+ if (flushThresholdTimeStr == null) {
+ // for backward compatibility with older property
+ key = StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME + StreamConfigProperties.LLC_SUFFIX;
+ flushThresholdTimeStr = streamConfigMap.get(key);
+ }
if (flushThresholdTimeStr != null) {
try {
return TimeUtils.convertPeriodToMillis(flushThresholdTimeStr);
@@ -286,8 +286,8 @@ public class StreamConfig {
// For backward-compatibility, parse it as milliseconds value
return Long.parseLong(flushThresholdTimeStr);
} catch (NumberFormatException nfe) {
- LOGGER.warn("Invalid config {}: {}, defaulting to: {}", StreamConfigProperties.SEGMENT_FLUSH_THRESHOLD_TIME,
- flushThresholdTimeStr, DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
+ LOGGER.warn("Invalid config {}: {}, defaulting to: {}", key, flushThresholdTimeStr,
+ DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS);
return DEFAULT_FLUSH_THRESHOLD_TIME_MILLIS;
}
}
@@ -304,16 +304,14 @@ public class StreamConfig {
return _topicName;
}
- public List<ConsumerType> getConsumerTypes() {
- return _consumerTypes;
- }
-
+ @Deprecated
public boolean hasHighLevelConsumerType() {
- return _consumerTypes.contains(ConsumerType.HIGHLEVEL);
+ return false;
}
+ @Deprecated
public boolean hasLowLevelConsumerType() {
- return _consumerTypes.contains(ConsumerType.LOWLEVEL);
+ return true;
}
public String getConsumerFactoryClassName() {
@@ -383,67 +381,46 @@ public class StreamConfig {
@Override
public String toString() {
- return "StreamConfig{" + "_type='" + _type + '\'' + ", _topicName='" + _topicName + '\'' + ", _consumerTypes="
- + _consumerTypes + ", _consumerFactoryClassName='" + _consumerFactoryClassName + '\'' + ", _offsetCriteria='"
- + _offsetCriteria + '\'' + ", _connectionTimeoutMillis=" + _connectionTimeoutMillis + ", _fetchTimeoutMillis="
- + _fetchTimeoutMillis + ", _idleTimeoutMillis=" + _idleTimeoutMillis + ", _flushThresholdRows="
- + _flushThresholdRows + ", _flushThresholdTimeMillis=" + _flushThresholdTimeMillis
- + ", _flushSegmentDesiredSizeBytes=" + _flushThresholdSegmentSizeBytes + ", _flushAutotuneInitialRows="
- + _flushAutotuneInitialRows + ", _decoderClass='" + _decoderClass + '\'' + ", _decoderProperties="
- + _decoderProperties + ", _groupId='" + _groupId + "', _topicConsumptionRateLimit=" + _topicConsumptionRateLimit
- + ", _tableNameWithType='" + _tableNameWithType + ", _serverUploadToDeepStore=" + _serverUploadToDeepStore
- + "}";
+ return "StreamConfig{" + "_type='" + _type + '\'' + ", _topicName='" + _topicName + '\'' + ", _tableNameWithType='"
+ + _tableNameWithType + '\'' + ", _consumerFactoryClassName='" + _consumerFactoryClassName + '\''
+ + ", _decoderClass='" + _decoderClass + '\'' + ", _decoderProperties=" + _decoderProperties
+ + ", _connectionTimeoutMillis=" + _connectionTimeoutMillis + ", _fetchTimeoutMillis=" + _fetchTimeoutMillis
+ + ", _idleTimeoutMillis=" + _idleTimeoutMillis + ", _flushThresholdRows=" + _flushThresholdRows
+ + ", _flushThresholdTimeMillis=" + _flushThresholdTimeMillis + ", _flushThresholdSegmentSizeBytes="
+ + _flushThresholdSegmentSizeBytes + ", _flushAutotuneInitialRows=" + _flushAutotuneInitialRows + ", _groupId='"
+ + _groupId + '\'' + ", _topicConsumptionRateLimit=" + _topicConsumptionRateLimit + ", _streamConfigMap="
+ + _streamConfigMap + ", _offsetCriteria=" + _offsetCriteria + ", _serverUploadToDeepStore="
+ + _serverUploadToDeepStore + '}';
}
@Override
public boolean equals(Object o) {
- if (EqualityUtils.isSameReference(this, o)) {
+ if (this == o) {
return true;
}
-
- if (EqualityUtils.isNullOrNotSameClass(this, o)) {
+ if (o == null || getClass() != o.getClass()) {
return false;
}
-
StreamConfig that = (StreamConfig) o;
-
- return EqualityUtils.isEqual(_connectionTimeoutMillis, that._connectionTimeoutMillis) && EqualityUtils.isEqual(
- _fetchTimeoutMillis, that._fetchTimeoutMillis) && EqualityUtils.isEqual(_idleTimeoutMillis,
- that._idleTimeoutMillis) && EqualityUtils.isEqual(_flushThresholdRows, that._flushThresholdRows)
- && EqualityUtils.isEqual(_flushThresholdTimeMillis, that._flushThresholdTimeMillis) && EqualityUtils.isEqual(
- _flushThresholdSegmentSizeBytes, that._flushThresholdSegmentSizeBytes) && EqualityUtils.isEqual(
- _flushAutotuneInitialRows, that._flushAutotuneInitialRows) && EqualityUtils.isEqual(_type, that._type)
- && EqualityUtils.isEqual(_topicName, that._topicName) && EqualityUtils.isEqual(_consumerTypes,
- that._consumerTypes) && EqualityUtils.isEqual(_consumerFactoryClassName, that._consumerFactoryClassName)
- && EqualityUtils.isEqual(_offsetCriteria, that._offsetCriteria) && EqualityUtils.isEqual(_decoderClass,
- that._decoderClass) && EqualityUtils.isEqual(_decoderProperties, that._decoderProperties)
- && EqualityUtils.isEqual(_groupId, that._groupId) && EqualityUtils.isEqual(_tableNameWithType,
- that._tableNameWithType) && EqualityUtils.isEqual(_topicConsumptionRateLimit, that._topicConsumptionRateLimit)
- && EqualityUtils.isEqual(_streamConfigMap, that._streamConfigMap)
- && _serverUploadToDeepStore == that._serverUploadToDeepStore;
+ return _connectionTimeoutMillis == that._connectionTimeoutMillis && _fetchTimeoutMillis == that._fetchTimeoutMillis
+ && _idleTimeoutMillis == that._idleTimeoutMillis && _flushThresholdRows == that._flushThresholdRows
+ && _flushThresholdTimeMillis == that._flushThresholdTimeMillis
+ && _flushThresholdSegmentSizeBytes == that._flushThresholdSegmentSizeBytes
+ && _flushAutotuneInitialRows == that._flushAutotuneInitialRows
+ && Double.compare(_topicConsumptionRateLimit, that._topicConsumptionRateLimit) == 0
+ && _serverUploadToDeepStore == that._serverUploadToDeepStore && Objects.equals(_type, that._type)
+ && Objects.equals(_topicName, that._topicName) && Objects.equals(_tableNameWithType, that._tableNameWithType)
+ && Objects.equals(_consumerFactoryClassName, that._consumerFactoryClassName) && Objects.equals(_decoderClass,
+ that._decoderClass) && Objects.equals(_decoderProperties, that._decoderProperties) && Objects.equals(_groupId,
+ that._groupId) && Objects.equals(_streamConfigMap, that._streamConfigMap) && Objects.equals(_offsetCriteria,
+ that._offsetCriteria);
}
@Override
public int hashCode() {
- int result = EqualityUtils.hashCodeOf(_type);
- result = EqualityUtils.hashCodeOf(result, _topicName);
- result = EqualityUtils.hashCodeOf(result, _consumerTypes);
- result = EqualityUtils.hashCodeOf(result, _consumerFactoryClassName);
- result = EqualityUtils.hashCodeOf(result, _offsetCriteria);
- result = EqualityUtils.hashCodeOf(result, _connectionTimeoutMillis);
- result = EqualityUtils.hashCodeOf(result, _fetchTimeoutMillis);
- result = EqualityUtils.hashCodeOf(result, _idleTimeoutMillis);
- result = EqualityUtils.hashCodeOf(result, _flushThresholdRows);
- result = EqualityUtils.hashCodeOf(result, _flushThresholdTimeMillis);
- result = EqualityUtils.hashCodeOf(result, _flushThresholdSegmentSizeBytes);
- result = EqualityUtils.hashCodeOf(result, _flushAutotuneInitialRows);
- result = EqualityUtils.hashCodeOf(result, _decoderClass);
- result = EqualityUtils.hashCodeOf(result, _decoderProperties);
- result = EqualityUtils.hashCodeOf(result, _groupId);
- result = EqualityUtils.hashCodeOf(result, _topicConsumptionRateLimit);
- result = EqualityUtils.hashCodeOf(result, _streamConfigMap);
- result = EqualityUtils.hashCodeOf(result, _tableNameWithType);
- result = EqualityUtils.hashCodeOf(result, _serverUploadToDeepStore);
- return result;
+ return Objects.hash(_type, _topicName, _tableNameWithType, _consumerFactoryClassName, _decoderClass,
+ _decoderProperties, _connectionTimeoutMillis, _fetchTimeoutMillis, _idleTimeoutMillis, _flushThresholdRows,
+ _flushThresholdTimeMillis, _flushThresholdSegmentSizeBytes, _flushAutotuneInitialRows, _groupId,
+ _topicConsumptionRateLimit, _streamConfigMap, _offsetCriteria, _serverUploadToDeepStore);
}
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
index b16984a301..df8dc87426 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConfigProperties.java
@@ -38,6 +38,7 @@ public class StreamConfigProperties {
*/
public static final String STREAM_TYPE = "streamType";
public static final String STREAM_TOPIC_NAME = "topic.name";
+ @Deprecated
public static final String STREAM_CONSUMER_TYPES = "consumer.type";
public static final String STREAM_CONSUMER_FACTORY_CLASS = "consumer.factory.class.name";
public static final String STREAM_CONSUMER_OFFSET_CRITERIA = "consumer.prop.auto.offset.reset";
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
index b4dc940315..9c23b571ea 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/builder/TableConfigBuilder.java
@@ -62,7 +62,6 @@ public class TableConfigBuilder {
private final TableType _tableType;
private String _tableName;
private boolean _isDimTable;
- private boolean _isLLC;
// Segments config related
private String _schemaName;
@@ -141,9 +140,10 @@ public class TableConfigBuilder {
return this;
}
+ @Deprecated
public TableConfigBuilder setLLC(boolean isLLC) {
Preconditions.checkState(_tableType == TableType.REALTIME);
- _isLLC = isLLC;
+ Preconditions.checkArgument(isLLC, "Real-time table must use LLC");
return this;
}
@@ -437,9 +437,6 @@ public class TableConfigBuilder {
validationConfig.setSchemaName(_schemaName);
validationConfig.setReplication(_numReplicas);
validationConfig.setPeerSegmentDownloadScheme(_peerSegmentDownloadScheme);
- if (_isLLC) {
- validationConfig.setReplicasPerPartition(_numReplicas);
- }
validationConfig.setCrypterClassName(_crypterClassName);
// Tenant config
diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
index 76fbdf4170..1b1d414249 100644
--- a/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/config/ConfigUtilsTest.java
@@ -59,7 +59,6 @@ public class ConfigUtilsTest {
String streamType = "fakeStream";
String topic = "fakeTopic";
- String consumerType = "simple";
String tableName = "fakeTable_REALTIME";
String defaultConsumerFactoryClass = "org.apache.pinot.plugin.stream.kafka20.StreamConsumerFactory";
String defaultDecoderClass = "org.apache.pinot.plugin.inputformat.avro.KafkaAvroMessageDecoder";
@@ -72,9 +71,6 @@ public class ConfigUtilsTest {
streamConfigMap
.put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_TOPIC_NAME),
topic);
- streamConfigMap
- .put(StreamConfigProperties.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_TYPES),
- consumerType);
streamConfigMap.put(StreamConfigProperties
.constructStreamProperty(streamType, StreamConfigProperties.STREAM_CONSUMER_FACTORY_CLASS),
consumerFactoryClass);
@@ -105,7 +101,6 @@ public class ConfigUtilsTest {
StreamConfig streamConfig = new StreamConfig(tableName, indexingConfig.getStreamConfigs());
Assert.assertEquals(streamConfig.getType(), streamType);
Assert.assertEquals(streamConfig.getTopicName(), topic);
- Assert.assertEquals(streamConfig.getConsumerTypes().get(0), StreamConfig.ConsumerType.LOWLEVEL);
Assert.assertEquals(streamConfig.getConsumerFactoryClassName(), defaultConsumerFactoryClass);
Assert.assertEquals(streamConfig.getDecoderClass(), defaultDecoderClass);
Assert.assertEquals(streamConfig.getStreamConfigsMap().get("stream.fakeStream.aws.accessKey"),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org