You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ne...@apache.org on 2021/01/02 23:52:17 UTC
[incubator-pinot] branch sharded_consumer_type_support created (now
0fa424e)
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a change to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at 0fa424e An attempt at server-side changes
This branch includes the following new commits:
new 54ce986 Controller side changes pseudo code
new 652a3e6 StreamPartitionOffset to implement Checkpoint
new 3ef1870 Rename partitionId to partitionGroupId
new b139460 Controller side code
new 1f81c1e Separate PartitionGroupInfo and PartitionGroupMetadata
new 1cb09cf More controller side changes
new eb0e4c7 Checnges in test to make it complie
new 0fa424e An attempt at server-side changes
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 08/08: An attempt at server-side changes
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 0fa424eaaeae3049e47d7f494cf86c7f88b84a05
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Thu Dec 31 17:19:24 2020 -0800
An attempt at server-side changes
---
.../realtime/LLRealtimeSegmentDataManager.java | 22 +++++++++++++---------
.../org/apache/pinot/spi/stream/FetchResult.java | 5 +----
.../pinot/spi/stream/PartitionGroupConsumer.java | 2 +-
3 files changed, 15 insertions(+), 14 deletions(-)
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 0cd1fba..dabe748 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
@@ -71,8 +71,10 @@ import org.apache.pinot.spi.config.table.SegmentPartitionConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.stream.FetchResult;
import org.apache.pinot.spi.stream.MessageBatch;
-import org.apache.pinot.spi.stream.PartitionLevelConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.PermanentConsumerException;
import org.apache.pinot.spi.stream.RowMetadata;
@@ -249,10 +251,11 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
private Thread _consumerThread;
private final String _streamTopic;
private final int _partitionGroupId;
+ private final PartitionGroupMetadata _partitionGroupMetadata;
final String _clientId;
private final LLCSegmentName _llcSegmentName;
private final RecordTransformer _recordTransformer;
- private PartitionLevelConsumer _partitionLevelConsumer = null;
+ private PartitionGroupConsumer _partitionGroupConsumer = null;
private StreamMetadataProvider _streamMetadataProvider = null;
private final File _resourceTmpDir;
private final String _tableNameWithType;
@@ -380,12 +383,10 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// Update _currentOffset upon return from this method
MessageBatch messageBatch;
try {
- messageBatch = _partitionLevelConsumer
+ FetchResult fetchResult = _partitionGroupConsumer
.fetchMessages(_currentOffset, null, _partitionLevelStreamConfig.getFetchTimeoutMillis());
+ messageBatch = fetchResult.getMessages();
consecutiveErrorCount = 0;
- } catch (TimeoutException e) {
- handleTransientStreamErrors(e);
- continue;
} catch (TransientConsumerException e) {
handleTransientStreamErrors(e);
continue;
@@ -898,7 +899,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
private void closePartitionLevelConsumer() {
try {
- _partitionLevelConsumer.close();
+ _partitionGroupConsumer.close();
} catch (Exception e) {
segmentLogger.warn("Could not close stream consumer", e);
}
@@ -1130,6 +1131,9 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_segmentNameStr = _segmentZKMetadata.getSegmentName();
_llcSegmentName = llcSegmentName;
_partitionGroupId = _llcSegmentName.getPartitionGroupId();
+ _partitionGroupMetadata = new PartitionGroupMetadata(_partitionGroupId, _llcSegmentName.getSequenceNumber(),
+ _segmentZKMetadata.getStartOffset(), _segmentZKMetadata.getEndOffset(),
+ _segmentZKMetadata.getStatus().toString());
_partitionGroupConsumerSemaphore = partitionGroupConsumerSemaphore;
_acquiredConsumerSemaphore = new AtomicBoolean(false);
_metricKeyName = _tableNameWithType + "-" + _streamTopic + "-" + _partitionGroupId;
@@ -1305,11 +1309,11 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
* @param reason
*/
private void makeStreamConsumer(String reason) {
- if (_partitionLevelConsumer != null) {
+ if (_partitionGroupConsumer != null) {
closePartitionLevelConsumer();
}
segmentLogger.info("Creating new stream consumer, reason: {}", reason);
- _partitionLevelConsumer = _streamConsumerFactory.createPartitionLevelConsumer(_clientId, _partitionGroupId);
+ _partitionGroupConsumer = _streamConsumerFactory.createPartitionGroupConsumer(_partitionGroupMetadata);
}
/**
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java
index b0ed6e5..7e8a911 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java
@@ -18,10 +18,7 @@
*/
package org.apache.pinot.spi.stream;
-import java.util.List;
-
-
public interface FetchResult<T> {
Checkpoint getLastCheckpoint();
- List<T> getMessages();
+ MessageBatch<T> getMessages();
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
index e096e67..bbbdaad 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
@@ -22,5 +22,5 @@ import java.io.Closeable;
public interface PartitionGroupConsumer extends Closeable {
- FetchResult fetch(Checkpoint start, Checkpoint end, long timeout);
+ FetchResult fetchMessages(Checkpoint start, Checkpoint end, long timeout);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 06/08: More controller side changes
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 1cb09cf59d2b4dd9b6ff79fdbe48775ae532ea60
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Thu Dec 31 15:39:44 2020 -0800
More controller side changes
---
.../helix/core/PinotHelixResourceManager.java | 4 +-
.../helix/core/PinotTableIdealStateBuilder.java | 16 +++---
.../realtime/PinotLLCRealtimeSegmentManager.java | 57 ++++++++--------------
.../PinotLLCRealtimeSegmentManagerTest.java | 4 +-
.../realtime/LLRealtimeSegmentDataManager.java | 2 +-
.../impl/fakestream/FakeStreamConsumerFactory.java | 2 +-
.../fakestream/FakeStreamMetadataProvider.java | 12 +++--
.../kafka09/KafkaStreamMetadataProvider.java | 36 ++++++++++----
.../kafka09/KafkaPartitionLevelConsumerTest.java | 4 +-
.../kafka20/KafkaStreamMetadataProvider.java | 3 +-
...Fetcher.java => PartitionGroupInfoFetcher.java} | 31 +++++-------
.../pinot/spi/stream/StreamMetadataProvider.java | 2 +-
12 files changed, 86 insertions(+), 87 deletions(-)
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 ebfbfa1..e80c06b 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
@@ -1337,7 +1337,7 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setupNewShardedTable(realtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setupNewTable(realtimeTableConfig, idealState);
LOGGER.info("Successfully setup table for SHARDED consumers for {} ", realtimeTableName);
} else {
@@ -1366,7 +1366,7 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setupNewShardedTable(realtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setupNewTable(realtimeTableConfig, idealState);
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);
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 a7b3c9e..8b200bb 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
@@ -30,10 +30,10 @@ import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.common.utils.StringUtil;
import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
-import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-import org.apache.pinot.spi.stream.PartitionGroupMetadataFetcher;
+import org.apache.pinot.spi.stream.PartitionGroupInfoFetcher;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -115,15 +115,15 @@ public class PinotTableIdealStateBuilder {
return idealState;
}
- public static List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig,
+ public static List<PartitionGroupInfo> getPartitionGroupInfoList(StreamConfig streamConfig,
List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
- PartitionGroupMetadataFetcher partitionGroupMetadataFetcher =
- new PartitionGroupMetadataFetcher(streamConfig, currentPartitionGroupMetadataList);
+ PartitionGroupInfoFetcher partitionGroupInfoFetcher =
+ new PartitionGroupInfoFetcher(streamConfig, currentPartitionGroupMetadataList);
try {
- RetryPolicies.noDelayRetryPolicy(3).attempt(partitionGroupMetadataFetcher);
- return partitionGroupMetadataFetcher.getPartitionGroupMetadataList();
+ RetryPolicies.noDelayRetryPolicy(3).attempt(partitionGroupInfoFetcher);
+ return partitionGroupInfoFetcher.getPartitionGroupInfoList();
} catch (Exception e) {
- Exception fetcherException = partitionGroupMetadataFetcher.getException();
+ Exception fetcherException = partitionGroupInfoFetcher.getException();
LOGGER.error("Could not get partition count for {}", streamConfig.getTopicName(), fetcherException);
throw new RuntimeException(fetcherException);
}
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 423a0b2..d899b4c 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
@@ -48,7 +48,6 @@ import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
import org.apache.pinot.common.metadata.segment.LLCRealtimeSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
-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.protocols.SegmentCompletionProtocol;
@@ -208,7 +207,7 @@ public class PinotLLCRealtimeSegmentManager {
/**
* Sets up the realtime table ideal state for a table of consumer type SHARDED
*/
- public void setupNewShardedTable(TableConfig tableConfig, IdealState idealState) {
+ public void setupNewTable(TableConfig tableConfig, IdealState idealState) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
@@ -220,18 +219,8 @@ public class PinotLLCRealtimeSegmentManager {
new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
// get new partition groups and their metadata
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
- .createStreamMetadataProvider(streamConfig.getTopicName() + "_" + System.currentTimeMillis());
-
- List<PartitionGroupInfo> newPartitionGroupMetadataList;
- try {
- newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000);
- } catch (TimeoutException e) {
- throw new IllegalStateException(e);
- }
- int numPartitionGroups = newPartitionGroupMetadataList.size();
+ List<PartitionGroupInfo> newPartitionGroupInfoList = getPartitionGroupInfoList(streamConfig, Collections.emptyList());
+ int numPartitionGroups = newPartitionGroupInfoList.size();
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
int numReplicas = getNumReplicas(tableConfig, instancePartitions);
@@ -242,7 +231,7 @@ public class PinotLLCRealtimeSegmentManager {
long currentTimeMs = getCurrentTimeMs();
Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
- for (PartitionGroupInfo partitionGroupInfo : newPartitionGroupMetadataList) {
+ for (PartitionGroupInfo partitionGroupInfo : newPartitionGroupInfoList) {
String segmentName = setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupInfo,
currentTimeMs, instancePartitions, numPartitionGroups, numReplicas);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment,
@@ -507,18 +496,10 @@ public class PinotLLCRealtimeSegmentManager {
List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
IngestionConfigUtils.getStreamConfigMap(tableConfig));
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
- .createStreamMetadataProvider(streamConfig.getTopicName() + " " + System.currentTimeMillis());
// find new partition groups [A],[B],[C],[D]
- List<PartitionGroupInfo> newPartitionGroupMetadataList;
- try {
- newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
- } catch (TimeoutException e) {
- throw new IllegalStateException(e);
- }
+ List<PartitionGroupInfo> newPartitionGroupMetadataList =
+ getPartitionGroupInfoList(streamConfig, currentPartitionGroupMetadataList);
// create new segment metadata, only if it is not IN_PROGRESS in the current state
Map<Integer, PartitionGroupMetadata> currentGroupIdToMetadata = currentPartitionGroupMetadataList.stream().collect(
@@ -721,9 +702,9 @@ public class PinotLLCRealtimeSegmentManager {
}
@VisibleForTesting
- List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig,
+ List<PartitionGroupInfo> getPartitionGroupInfoList(StreamConfig streamConfig,
List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
- return PinotTableIdealStateBuilder.getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList);
+ return PinotTableIdealStateBuilder.getPartitionGroupInfoList(streamConfig, currentPartitionGroupMetadataList);
}
@VisibleForTesting
@@ -843,7 +824,7 @@ public class PinotLLCRealtimeSegmentManager {
if (idealState.isEnabled()) {
List<PartitionGroupMetadata> currentPartitionGroupMetadataList =
getCurrentPartitionGroupMetadataList(idealState);
- int numPartitions = getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList).size();
+ int numPartitions = getPartitionGroupInfoList(streamConfig, currentPartitionGroupMetadataList).size();
return ensureAllPartitionsConsuming(tableConfig, streamConfig, idealState, numPartitions);
} else {
LOGGER.info("Skipping LLC segments validation for disabled table: {}", realtimeTableName);
@@ -1005,7 +986,7 @@ public class PinotLLCRealtimeSegmentManager {
// and restart consumption from the same offset (if possible) or a newer offset (if realtime stream does not have the same offset).
// In latter case, report data loss.
for (Map.Entry<Integer, LLCRealtimeSegmentZKMetadata> entry : latestSegmentZKMetadataMap.entrySet()) {
- int partitionId = entry.getKey();
+ int partitionGroupId = entry.getKey();
LLCRealtimeSegmentZKMetadata latestSegmentZKMetadata = entry.getValue();
String latestSegmentName = latestSegmentZKMetadata.getSegmentName();
LLCSegmentName latestLLCSegmentName = new LLCSegmentName(latestSegmentName);
@@ -1049,10 +1030,10 @@ public class PinotLLCRealtimeSegmentManager {
StreamPartitionMsgOffset startOffset = offsetFactory.create(latestSegmentZKMetadata.getStartOffset());
// Start offset must be higher than the start offset of the stream
StreamPartitionMsgOffset partitionStartOffset =
- getPartitionOffset(streamConfig, OffsetCriteria.SMALLEST_OFFSET_CRITERIA, partitionId);
+ getPartitionOffset(streamConfig, OffsetCriteria.SMALLEST_OFFSET_CRITERIA, partitionGroupId);
if (partitionStartOffset.compareTo(startOffset) > 0) {
LOGGER.error("Data lost from offset: {} to: {} for partition: {} of table: {}", startOffset,
- partitionStartOffset, partitionId, realtimeTableName);
+ partitionStartOffset, partitionGroupId, realtimeTableName);
_controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
startOffset = partitionStartOffset;
}
@@ -1089,7 +1070,7 @@ public class PinotLLCRealtimeSegmentManager {
String previousConsumingSegment = null;
for (Map.Entry<String, Map<String, String>> segmentEntry : instanceStatesMap.entrySet()) {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentEntry.getKey());
- if (llcSegmentName.getPartitionGroupId() == partitionId && segmentEntry.getValue()
+ if (llcSegmentName.getPartitionGroupId() == partitionGroupId && segmentEntry.getValue()
.containsValue(SegmentStateModel.CONSUMING)) {
previousConsumingSegment = llcSegmentName.getSegmentName();
break;
@@ -1098,7 +1079,7 @@ public class PinotLLCRealtimeSegmentManager {
if (previousConsumingSegment == null) {
LOGGER
.error("Failed to find previous CONSUMING segment for partition: {} of table: {}, potential data loss",
- partitionId, realtimeTableName);
+ partitionGroupId, realtimeTableName);
_controllerMetrics.addMeteredTableValue(realtimeTableName, ControllerMeter.LLC_STREAM_DATA_LOSS, 1L);
}
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, previousConsumingSegment, latestSegmentName,
@@ -1111,10 +1092,14 @@ public class PinotLLCRealtimeSegmentManager {
}
// Set up new partitions if not exist
- for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
- if (!latestSegmentZKMetadataMap.containsKey(partitionId)) {
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
+ List<PartitionGroupInfo> partitionGroupInfoList =
+ getPartitionGroupInfoList(streamConfig, currentPartitionGroupMetadataList);
+ for (PartitionGroupInfo partitionGroupInfo : partitionGroupInfoList) {
+ int partitionGroupId = partitionGroupInfo.getPartitionGroupId();
+ if (!latestSegmentZKMetadataMap.containsKey(partitionGroupId)) {
String newSegmentName =
- setupNewPartitionGroup(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions,
+ setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupInfo, currentTimeMs, instancePartitions, numPartitions,
numReplicas);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, newSegmentName, segmentAssignment,
instancePartitionsMap);
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 743e719..42bdedc 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
@@ -50,7 +50,6 @@ import org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignme
import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import org.apache.pinot.controller.util.SegmentCompletionUtils;
import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
-import org.apache.pinot.core.realtime.impl.fakestream.FakePartitionGroupMetadata;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.core.segment.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -60,6 +59,7 @@ import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.StreamConfig;
@@ -914,7 +914,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
@Override
- List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ List<PartitionGroupInfo> getPartitionGroupInfoList(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
return IntStream.range(0, _numPartitions).mapToObj(FakePartitionGroupMetadata::new).collect(Collectors.toList());
}
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 f00facc..0cd1fba 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
@@ -1239,7 +1239,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
int numPartitions = columnPartitionConfig.getNumPartitions();
try {
// fixme: get this from ideal state
- int numStreamPartitions = _streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000).size();
+ int numStreamPartitions = _streamMetadataProvider.getPartitionGroupInfoList(Collections.emptyList(), 5000).size();
if (numStreamPartitions != numPartitions) {
segmentLogger.warn(
"Number of stream partitions: {} does not match number of partitions in the partition config: {}, using number of stream partitions",
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
index 54be1b6..6121eef 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
@@ -88,7 +88,7 @@ public class FakeStreamConsumerFactory extends StreamConsumerFactory {
// stream metadata provider
StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(clientId);
- int partitionCount = streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 10_000).size();
+ int partitionCount = streamMetadataProvider.getPartitionGroupInfoList(Collections.emptyList(), 10_000).size();
System.out.println(partitionCount);
// Partition metadata provider
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
index c96d06a..0de0ce2 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nonnull;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
@@ -35,9 +36,11 @@ import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
*/
public class FakeStreamMetadataProvider implements StreamMetadataProvider {
private final int _numPartitions;
+ private StreamConfig _streamConfig;
public FakeStreamMetadataProvider(StreamConfig streamConfig) {
_numPartitions = FakeStreamConfigUtils.getNumPartitions(streamConfig);
+ _streamConfig = streamConfig;
}
@Override
@@ -46,11 +49,12 @@ public class FakeStreamMetadataProvider implements StreamMetadataProvider {
}
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
- List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
+ public List<PartitionGroupInfo> getPartitionGroupInfoList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
+ throws TimeoutException {
+ List<PartitionGroupInfo> partitionGroupMetadataList = new ArrayList<>();
for (int i = 0; i < _numPartitions; i++) {
- partitionGroupMetadataList.add(new FakePartitionGroupMetadata(i));
+ partitionGroupMetadataList.add(new PartitionGroupInfo(i, fetchStreamPartitionOffset(_streamConfig.getOffsetCriteria(), 5000).toString()));
}
return partitionGroupMetadataList;
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
index 865ae96..2d0ad31 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
@@ -40,6 +40,7 @@ import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
@@ -54,13 +55,14 @@ import org.slf4j.LoggerFactory;
public class KafkaStreamMetadataProvider extends KafkaConnectionHandler implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private StreamConfig _streamConfig;
+
/**
* Create a partition specific metadata provider
- * @param streamConfig
- * @param partition
*/
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition, new KafkaSimpleConsumerFactoryImpl());
+ _streamConfig = streamConfig;
}
/**
@@ -69,18 +71,21 @@ public class KafkaStreamMetadataProvider extends KafkaConnectionHandler implemen
*/
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
super(clientId, streamConfig, new KafkaSimpleConsumerFactoryImpl());
+ _streamConfig = streamConfig;
}
@VisibleForTesting
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition,
KafkaSimpleConsumerFactory kafkaSimpleConsumerFactory) {
super(clientId, streamConfig, partition, kafkaSimpleConsumerFactory);
+ _streamConfig = streamConfig;
}
@VisibleForTesting
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig,
KafkaSimpleConsumerFactory kafkaSimpleConsumerFactory) {
super(clientId, streamConfig, kafkaSimpleConsumerFactory);
+ _streamConfig = streamConfig;
}
/**
@@ -156,19 +161,30 @@ public class KafkaStreamMetadataProvider extends KafkaConnectionHandler implemen
}
/**
- * Fetch the partition group metadata list
+ * Fetch the partitionGroupMetadata list.
* @param currentPartitionGroupsMetadata In case of Kafka, each partition group contains a single partition.
- * Hence current partition groups are not needed to compute the new partition groups
*/
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- @Nullable List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
+ public List<PartitionGroupInfo> getPartitionGroupInfoList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
+ throws java.util.concurrent.TimeoutException {
int partitionCount = fetchPartitionCountInternal(timeoutMillis);
- List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>(partitionCount);
- for (int i = 0; i < partitionCount; i++) {
- partitionGroupMetadataList.add(new KafkaPartitionGroupMetadata(i));
+ List<PartitionGroupInfo> newPartitionGroupInfoList = new ArrayList<>(partitionCount);
+
+ // add a PartitionGroupInfo into the list foreach partition already present in current.
+ // the end checkpoint is set as checkpoint
+ for (PartitionGroupMetadata currentPartitionGroupMetadata : currentPartitionGroupsMetadata) {
+ newPartitionGroupInfoList.add(new PartitionGroupInfo(currentPartitionGroupMetadata.getPartitionGroupId(),
+ currentPartitionGroupMetadata.getEndCheckpoint()));
+ }
+ // add PartitiongroupInfo for new partitions
+ // use offset criteria from stream config
+ for (int i = currentPartitionGroupsMetadata.size(); i < partitionCount; i++) {
+ StreamPartitionMsgOffset streamPartitionMsgOffset =
+ fetchStreamPartitionOffset(_streamConfig.getOffsetCriteria(), 5000);
+ newPartitionGroupInfoList.add(new PartitionGroupInfo(i, streamPartitionMsgOffset.toString()));
}
- return partitionGroupMetadataList;
+ return newPartitionGroupInfoList;
}
public synchronized long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis)
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
index 43b72a8..9d3091e 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
@@ -267,7 +267,7 @@ public class KafkaPartitionLevelConsumerTest {
}
@Test
- public void testGetPartitionCount() {
+ public void testGetPartitionCount() throws Exception {
String streamType = "kafka";
String streamKafkaTopicName = "theTopic";
String streamKafkaBrokerList = "abcd:1234,bcde:2345";
@@ -291,7 +291,7 @@ public class KafkaPartitionLevelConsumerTest {
KafkaStreamMetadataProvider streamMetadataProvider =
new KafkaStreamMetadataProvider(clientId, streamConfig, mockKafkaSimpleConsumerFactory);
- Assert.assertEquals(streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 10000L), 2);
+ Assert.assertEquals(streamMetadataProvider.getPartitionGroupInfoList(Collections.emptyList(), 10000L), 2);
}
@Test
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
index eb606f2..ef22b6a 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
@@ -58,10 +58,9 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
/**
* Fetch the partitionGroupMetadata list.
* @param currentPartitionGroupsMetadata In case of Kafka, each partition group contains a single partition.
- * Hence current partition groups are not needed to compute the new partition groups
*/
@Override
- public List<PartitionGroupInfo> getPartitionGroupMetadataList(
+ public List<PartitionGroupInfo> getPartitionGroupInfoList(
List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
throws TimeoutException {
int partitionCount = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis)).size();
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfoFetcher.java
similarity index 64%
rename from pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
rename to pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfoFetcher.java
index e1ce1a6..d13be10 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfoFetcher.java
@@ -27,27 +27,24 @@ import org.slf4j.LoggerFactory;
/**
* Fetches the partition count of a stream using the {@link StreamMetadataProvider}
*/
-public class PartitionGroupMetadataFetcher implements Callable<Boolean> {
+public class PartitionGroupInfoFetcher implements Callable<Boolean> {
- private static final Logger LOGGER = LoggerFactory.getLogger(PartitionGroupMetadataFetcher.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionGroupInfoFetcher.class);
- private int _partitionCount = -1;
- private List<PartitionGroupMetadata> _partitionGroupMetadataList;
- private List<PartitionGroupMetadata> _currentPartitionGroupMetadata;
- private final StreamConfig _streamConfig;
- private StreamConsumerFactory _streamConsumerFactory;
+ private List<PartitionGroupInfo> _partitionGroupInfoList;
+ private final List<PartitionGroupMetadata> _currentPartitionGroupMetadata;
+ private final StreamConsumerFactory _streamConsumerFactory;
private Exception _exception;
private final String _topicName;
- public PartitionGroupMetadataFetcher(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
- _streamConfig = streamConfig;
- _streamConsumerFactory = StreamConsumerFactoryProvider.create(_streamConfig);
+ public PartitionGroupInfoFetcher(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ _streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
_topicName = streamConfig.getTopicName();
_currentPartitionGroupMetadata = currentPartitionGroupMetadataList;
}
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList() {
- return _partitionGroupMetadataList;
+ public List<PartitionGroupInfo> getPartitionGroupInfoList() {
+ return _partitionGroupInfoList;
}
public Exception getException() {
@@ -55,21 +52,19 @@ public class PartitionGroupMetadataFetcher implements Callable<Boolean> {
}
/**
- * Callable to fetch the number of partitions of the stream given the stream metadata
- * @return
- * @throws Exception
+ * Callable to fetch the partition group info for the stream
*/
@Override
public Boolean call()
throws Exception {
- String clientId = PartitionGroupMetadataFetcher.class.getSimpleName() + "-" + _topicName;
+ String clientId = PartitionGroupInfoFetcher.class.getSimpleName() + "-" + _topicName;
try (
StreamMetadataProvider streamMetadataProvider = _streamConsumerFactory.createStreamMetadataProvider(clientId)) {
- _partitionGroupMetadataList = streamMetadataProvider.getPartitionGroupMetadataList(_currentPartitionGroupMetadata, /*maxWaitTimeMs=*/5000L);
+ _partitionGroupInfoList = streamMetadataProvider.getPartitionGroupInfoList(_currentPartitionGroupMetadata, /*maxWaitTimeMs=*/5000L);
if (_exception != null) {
// We had at least one failure, but succeeded now. Log an info
- LOGGER.info("Successfully retrieved partition count as {} for topic {}", _partitionCount, _topicName);
+ LOGGER.info("Successfully retrieved partition group info for topic {}", _topicName);
}
return Boolean.TRUE;
} catch (TransientConsumerException e) {
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
index a9cd2d6..f595ea3 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
@@ -40,7 +40,7 @@ public interface StreamMetadataProvider extends Closeable {
int fetchPartitionCount(long timeoutMillis);
// takes the current state of partition groups (groupings of shards, the state of the consumption) and creates the new state
- List<PartitionGroupInfo> getPartitionGroupMetadataList(
+ List<PartitionGroupInfo> getPartitionGroupInfoList(
List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
throws TimeoutException;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 07/08: Checnges in test to make it complie
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit eb0e4c782829dd3d33ae50ee91def7e09028b090
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Thu Dec 31 15:49:33 2020 -0800
Checnges in test to make it complie
---
.../controller/helix/core/PinotHelixResourceManager.java | 4 ++--
.../core/realtime/PinotLLCRealtimeSegmentManager.java | 2 +-
.../core/realtime/PinotLLCRealtimeSegmentManagerTest.java | 15 +++++++++++----
3 files changed, 14 insertions(+), 7 deletions(-)
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 e80c06b..5388eeb 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
@@ -1337,7 +1337,7 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setupNewTable(realtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
LOGGER.info("Successfully setup table for SHARDED consumers for {} ", realtimeTableName);
} else {
@@ -1366,7 +1366,7 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setupNewTable(realtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
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);
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 d899b4c..cf3a401 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
@@ -207,7 +207,7 @@ public class PinotLLCRealtimeSegmentManager {
/**
* Sets up the realtime table ideal state for a table of consumer type SHARDED
*/
- public void setupNewTable(TableConfig tableConfig, IdealState idealState) {
+ public void setUpNewTable(TableConfig tableConfig, IdealState idealState) {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
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 42bdedc..75c8057 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
@@ -907,15 +907,22 @@ public class PinotLLCRealtimeSegmentManagerTest {
@Override
void updateIdealStateOnSegmentCompletion(String realtimeTableName, String committingSegmentName,
- String newSegmentName, SegmentAssignment segmentAssignment,
+ List<String> newSegmentNames, SegmentAssignment segmentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
updateInstanceStatesForNewConsumingSegment(_idealState.getRecord().getMapFields(), committingSegmentName,
- newSegmentName, segmentAssignment, instancePartitionsMap);
+ null, segmentAssignment, instancePartitionsMap);
+ for (String segmentName : newSegmentNames) {
+ updateInstanceStatesForNewConsumingSegment(_idealState.getRecord().getMapFields(), null,
+ segmentName, segmentAssignment, instancePartitionsMap);
+ }
}
@Override
- List<PartitionGroupInfo> getPartitionGroupInfoList(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
- return IntStream.range(0, _numPartitions).mapToObj(FakePartitionGroupMetadata::new).collect(Collectors.toList());
+ List<PartitionGroupInfo> getPartitionGroupInfoList(StreamConfig streamConfig,
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ return IntStream.range(0, _numPartitions).mapToObj(i -> new PartitionGroupInfo(i,
+ getPartitionOffset(streamConfig, OffsetCriteria.SMALLEST_OFFSET_CRITERIA, i).toString()))
+ .collect(Collectors.toList());
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/08: Controller side changes pseudo code
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 54ce986d49eb418a949c04d7d9832f81e4453d65
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Wed Dec 23 17:08:08 2020 -0800
Controller side changes pseudo code
---
.../helix/core/PinotHelixResourceManager.java | 29 ++++++++
.../realtime/PinotLLCRealtimeSegmentManager.java | 79 ++++++++++++++++++++++
.../org/apache/pinot/spi/stream/Checkpoint.java | 24 +++++++
.../org/apache/pinot/spi/stream/FetchResult.java | 27 ++++++++
.../pinot/spi/stream/PartitionGroupConsumer.java | 23 +++++++
.../pinot/spi/stream/PartitionGroupMetadata.java | 41 +++++++++++
.../spi/stream/PartitionGroupMetadataList.java | 30 ++++++++
.../org/apache/pinot/spi/stream/StreamConfig.java | 6 +-
.../pinot/spi/stream/StreamConsumerFactory.java | 9 ++-
9 files changed, 266 insertions(+), 2 deletions(-)
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 6b5168f..5afbe7e 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
@@ -85,6 +85,7 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants.Helix;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.BrokerResourceStateModel;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
@@ -122,7 +123,10 @@ import org.apache.pinot.spi.config.table.TenantConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
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.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -1332,6 +1336,10 @@ public class PinotHelixResourceManager {
IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
IdealState idealState = getTableIdealState(realtimeTableName);
+ if (streamConfig.isShardedConsumerType()) {
+ setupShardedRealtimeTable(streamConfig, idealState, realtimeTableConfig.getValidationConfig().getReplicasPerPartitionNumber());
+ }
+
if (streamConfig.hasHighLevelConsumerType()) {
if (idealState == null) {
LOGGER.info("Initializing IdealState for HLC table: {}", realtimeTableName);
@@ -1364,6 +1372,27 @@ public class PinotHelixResourceManager {
}
}
+ /**
+ * Sets up the realtime table ideal state
+ * @param streamConfig
+ */
+ private void setupShardedRealtimeTable(StreamConfig streamConfig, IdealState idealState, int numReplicas) {
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+
+ // get current partition groups and their metadata - this will be empty when creating the table
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList = _pinotLLCRealtimeSegmentManager.getCurrentPartitionGroupMetadataList(idealState);
+
+ // get new partition groups and their metadata,
+ // Assume table has 3 shards. Say we get [0], [1], [2] groups (for now assume that each group contains only 1 shard)
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList =
+ streamConsumerFactory.getPartitionGroupMetadataList(currentPartitionGroupMetadataList);
+
+ // setup segment zk metadata and ideal state for all the new found partition groups
+ _pinotLLCRealtimeSegmentManager.setupNewPartitionGroups(newPartitionGroupMetadataList, numReplicas);
+ }
+
+
+
private void ensurePropertyStoreEntryExistsForHighLevelConsumer(String realtimeTableName) {
String propertyStorePath = ZKMetadataProvider.constructPropertyStorePathForResource(realtimeTableName);
if (!_propertyStore.exists(propertyStorePath, AccessOption.PERSISTENT)) {
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 00433b3..b459760 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
@@ -45,6 +45,7 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
import org.apache.pinot.common.metadata.segment.LLCRealtimeSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
+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.protocols.SegmentCompletionProtocol;
@@ -75,10 +76,12 @@ import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.PartitionOffsetFetcher;
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.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
@@ -157,6 +160,82 @@ public class PinotLLCRealtimeSegmentManager {
_flushThresholdUpdateManager = new FlushThresholdUpdateManager();
}
+ /**
+ * The committing segment will call this.
+ *
+ * For example, say we have 3 shards, grouped into PartitionGroups as [0], [1], [2]
+ * Now segment of PG (partition group) 0 is committing. First, we'll update the metadata to DONE, and ideal state to ONLINE
+ * Then, the currentPartitionGroupMetadata list will contain - [1], [2]
+ * The newPartitionGroupMetadata list will contain - [0], [1], [2]
+ * We then get the set of PGs for which new segments need to be made - [0]
+ */
+ public void commitPartitionGroup(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) {
+ TableConfig realtimeTableConfig = getTableConfig(realtimeTableName);
+ StreamConfig streamConfig = new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
+ int numReplicas = realtimeTableConfig.getValidationConfig().getReplicasPerPartitionNumber();
+ IdealState idealState = getIdealState(realtimeTableName);
+
+ // update status in segment metadata to DONE
+ // ..
+
+ // update Ideal State for this segment to ONLINE
+ // ..
+
+ // fetch current partition groups (which are actively CONSUMING - from example above, [1], [2])
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
+
+ // get new partition groups (honor any groupings which are already consuming - [0], [1], [2])
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList =
+ streamConsumerFactory.getPartitionGroupMetadataList(currentPartitionGroupMetadataList);
+
+ // from the above list, remove the partition groups which are already CONSUMING
+ // i.e. newPartitionGroups - currentPartitionGroups. Therefore, ([0], [1], [2]) - ([1], [2]) = ([0])
+ // ..
+
+ // setup segment metadata and ideal state for the new found partition groups
+ setupNewPartitionGroups(newPartitionGroupMetadataList, numReplicas);
+ }
+
+ public void setupIdealStateForConsuming(List<SegmentZKMetadata> segmentZKMetadata, int numReplicas) {
+ // add all segments from the list to ideal state, with state CONSUMING
+ }
+
+ public void persistSegmentMetadata(List<SegmentZKMetadata> segmentMetadata) {
+ // persist new segment metadata from list to zk
+ }
+
+ /**
+ * Using the list of partition group metadata, create a list of equivalent segment zk metadata
+ */
+ public List<SegmentZKMetadata> constructSegmentMetadata(List<PartitionGroupMetadata> partitionGroupMetadataList) {
+ List<SegmentZKMetadata> segmentZKMetadata = new ArrayList<>();
+ // for each partition group construct a segment zk metadata object
+ return segmentZKMetadata;
+ }
+
+ /**
+ * Using the ideal state, return a list of the current partition groups
+ */
+ public List<PartitionGroupMetadata> getCurrentPartitionGroupMetadataList(IdealState idealState) {
+ List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
+ // from all segment names in the ideal state, find unique groups
+
+ // create a PartitionGroupMetadata, one for each group
+ return partitionGroupMetadataList;
+ }
+
+ public void setupNewPartitionGroups(List<PartitionGroupMetadata> newPartitionGroupMetadataList, int numReplicas) {
+ // construct segment zk metadata for the new partition groups
+ List<SegmentZKMetadata> segmentMetadata = constructSegmentMetadata(newPartitionGroupMetadataList);
+
+ // create these new segments metadata
+ persistSegmentMetadata(segmentMetadata);
+
+ // setup ideal state for the new segments
+ setupIdealStateForConsuming(segmentMetadata, numReplicas);
+ }
+
public boolean getIsSplitCommitEnabled() {
return _controllerConf.getAcceptSplitCommit();
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java
new file mode 100644
index 0000000..627c964
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java
@@ -0,0 +1,24 @@
+/**
+ * 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.spi.stream;
+
+public interface Checkpoint {
+ byte[] serialize();
+ Checkpoint deserialize(byte[] blob);
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java
new file mode 100644
index 0000000..b0ed6e5
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/FetchResult.java
@@ -0,0 +1,27 @@
+/**
+ * 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.spi.stream;
+
+import java.util.List;
+
+
+public interface FetchResult<T> {
+ Checkpoint getLastCheckpoint();
+ List<T> getMessages();
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
new file mode 100644
index 0000000..2f138c2
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
@@ -0,0 +1,23 @@
+/**
+ * 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.spi.stream;
+
+public interface PartitionGroupConsumer {
+ FetchResult fetch(Checkpoint start, Checkpoint end, long timeout);
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
new file mode 100644
index 0000000..779c167
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
@@ -0,0 +1,41 @@
+/**
+ * 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.spi.stream;
+
+import java.util.List;
+
+
+public interface PartitionGroupMetadata {
+
+ int getGroupId();
+
+ List<String> getPartitions();
+
+ Checkpoint getStartCheckpoint(); // similar to getStartOffset
+
+ Checkpoint getEndCheckpoint(); // similar to getEndOffset
+
+ void setStartCheckpoint(Checkpoint startCheckpoint);
+
+ void setEndCheckpoint(Checkpoint endCheckpoint);
+
+ byte[] serialize();
+
+ PartitionGroupMetadata deserialize(byte[] blob);
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataList.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataList.java
new file mode 100644
index 0000000..1568d63
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataList.java
@@ -0,0 +1,30 @@
+/**
+ * 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.spi.stream;
+
+import java.util.List;
+
+
+public interface PartitionGroupMetadataList {
+
+ List<PartitionGroupMetadata> getMetadataList();
+
+ PartitionGroupMetadata getPartitionGroupMetadata(int index);
+
+}
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 d343203..a3e359e 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
@@ -41,7 +41,7 @@ public class StreamConfig {
* 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
+ HIGHLEVEL, LOWLEVEL, SHARDED
}
public static final int DEFAULT_FLUSH_THRESHOLD_ROWS = 5_000_000;
@@ -273,6 +273,10 @@ public class StreamConfig {
return _consumerTypes.contains(ConsumerType.LOWLEVEL);
}
+ public boolean isShardedConsumerType() {
+ return _consumerTypes.size() == 1 && _consumerTypes.get(0).equals(ConsumerType.SHARDED);
+ }
+
public String getConsumerFactoryClassName() {
return _consumerFactoryClassName;
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
index 27205c9..4db0fb1 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
@@ -18,8 +18,8 @@
*/
package org.apache.pinot.spi.stream;
+import java.util.List;
import java.util.Set;
-import org.apache.pinot.spi.data.Schema;
/**
@@ -73,4 +73,11 @@ public abstract class StreamConsumerFactory {
public StreamPartitionMsgOffsetFactory createStreamMsgOffsetFactory() {
return new LongMsgOffsetFactory();
}
+
+ // takes the current state of partition groups (groupings of shards, the state of the consumption) and creates the new state
+ public abstract List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata);
+
+ // creates a consumer which consumes from a partition group
+ public abstract PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 04/08: Controller side code
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit b13946066ce3e30375f18ddabd4023c113219eb6
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Thu Dec 31 12:04:54 2020 -0800
Controller side code
---
.../segment/RealtimeSegmentZKMetadata.java | 6 -
.../helix/core/PinotHelixResourceManager.java | 89 ++++-----
.../helix/core/PinotTableIdealStateBuilder.java | 9 +-
.../realtime/PinotLLCRealtimeSegmentManager.java | 201 +++++++++++++--------
.../fakestream/FakePartitionGroupMetadata.java | 48 -----
.../kafka09/KafkaPartitionGroupMetadata.java | 48 -----
.../kafka20/KafkaPartitionGroupMetadata.java | 48 -----
.../pinot/spi/stream/PartitionGroupMetadata.java | 52 +++++-
8 files changed, 207 insertions(+), 294 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java b/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
index c46af53..d88be18 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
@@ -35,7 +35,6 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
private Status _status = null;
private int _sizeThresholdToFlushSegment = -1;
private String _timeThresholdToFlushSegment = null; // store as period string for readability
- private String _partitionGroupMetadataStr = null;
public RealtimeSegmentZKMetadata() {
setSegmentType(SegmentType.REALTIME);
@@ -50,7 +49,6 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
if (flushThresholdTime != null && !flushThresholdTime.equals(NULL)) {
_timeThresholdToFlushSegment = znRecord.getSimpleField(CommonConstants.Segment.FLUSH_THRESHOLD_TIME);
}
- _partitionGroupMetadataStr = znRecord.getSimpleField(CommonConstants.Segment.PARTITION_GROUP_METADATA);
}
@Override
@@ -143,8 +141,4 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
public void setTimeThresholdToFlushSegment(String timeThresholdPeriodString) {
_timeThresholdToFlushSegment = timeThresholdPeriodString;
}
-
- public String getPartitionGroupMetadataStr() {
- return _partitionGroupMetadataStr;
- }
}
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 7f36e5a..1ff3f9d 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
@@ -85,7 +85,6 @@ import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
import org.apache.pinot.common.metadata.segment.OfflineSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
-import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.utils.CommonConstants.Helix;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.BrokerResourceStateModel;
import org.apache.pinot.common.utils.CommonConstants.Helix.StateModel.SegmentStateModel;
@@ -124,6 +123,7 @@ import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.data.Schema;
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.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
@@ -1337,65 +1337,50 @@ public class PinotHelixResourceManager {
IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
IdealState idealState = getTableIdealState(realtimeTableName);
+
if (streamConfig.isShardedConsumerType()) {
- setupShardedRealtimeTable(streamConfig, idealState, realtimeTableConfig.getValidationConfig().getReplicasPerPartitionNumber());
- }
+ idealState = PinotTableIdealStateBuilder
+ .buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
+ _enableBatchMessageMode);
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
+ LOGGER.info("Successfully added Helix entries for low-level consumers for {} ", realtimeTableName);
+ _pinotLLCRealtimeSegmentManager.setupNewShardedTable(rawRealtimeTableConfig, idealState);
+ } else {
- 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);
+ 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);
}
- // 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);
+
+ // 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()) {
+ idealState = PinotTableIdealStateBuilder
+ .buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
+ _enableBatchMessageMode);
+ _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
+ 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);
+ }
}
}
}
- /**
- * Sets up the realtime table ideal state
- * @param streamConfig
- */
- private void setupShardedRealtimeTable(StreamConfig streamConfig, IdealState idealState, int numReplicas) {
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
- .createStreamMetadataProvider(streamConfig.getTopicName() + "_" + System.currentTimeMillis());
-
- // get current partition groups and their metadata - this will be empty when creating the table
- List<PartitionGroupMetadata> currentPartitionGroupMetadataList = _pinotLLCRealtimeSegmentManager.getCurrentPartitionGroupMetadataList(idealState);
-
- // get new partition groups and their metadata,
- // Assume table has 3 shards. Say we get [0], [1], [2] groups (for now assume that each group contains only 1 shard)
- List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 5000);
-
- // setup segment zk metadata and ideal state for all the new found partition groups
- _pinotLLCRealtimeSegmentManager.setupNewPartitionGroups(newPartitionGroupMetadataList, numReplicas);
- }
-
-
-
private void ensurePropertyStoreEntryExistsForHighLevelConsumer(String realtimeTableName) {
String propertyStorePath = ZKMetadataProvider.constructPropertyStorePathForResource(realtimeTableName);
if (!_propertyStore.exists(propertyStorePath, AccessOption.PERSISTENT)) {
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 1e95966..a7b3c9e 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
@@ -94,9 +94,8 @@ public class PinotTableIdealStateBuilder {
return idealState;
}
- public static void buildLowLevelRealtimeIdealStateFor(PinotLLCRealtimeSegmentManager pinotLLCRealtimeSegmentManager,
- String realtimeTableName, TableConfig realtimeTableConfig, IdealState idealState,
- boolean enableBatchMessageMode) {
+ public static IdealState buildLowLevelRealtimeIdealStateFor(String realtimeTableName, TableConfig realtimeTableConfig,
+ IdealState idealState, boolean enableBatchMessageMode) {
// Validate replicasPerPartition here.
final String replicasPerPartitionStr = realtimeTableConfig.getValidationConfig().getReplicasPerPartition();
@@ -105,7 +104,7 @@ public class PinotTableIdealStateBuilder {
}
final int nReplicas;
try {
- nReplicas = Integer.valueOf(replicasPerPartitionStr);
+ nReplicas = Integer.parseInt(replicasPerPartitionStr);
} catch (NumberFormatException e) {
throw new PinotHelixResourceManager.InvalidTableConfigException(
"Invalid value for replicasPerPartition, expected a number: " + replicasPerPartitionStr, e);
@@ -113,7 +112,7 @@ public class PinotTableIdealStateBuilder {
if (idealState == null) {
idealState = buildEmptyRealtimeIdealStateFor(realtimeTableName, nReplicas, enableBatchMessageMode);
}
- pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
+ return idealState;
}
public static List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig,
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 6b220f2..0de5e91 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
@@ -32,6 +32,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.helix.AccessOption;
import org.apache.helix.HelixAdmin;
@@ -44,6 +45,7 @@ import org.apache.pinot.common.assignment.InstancePartitionsUtils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.ColumnPartitionMetadata;
import org.apache.pinot.common.metadata.segment.LLCRealtimeSegmentZKMetadata;
+import org.apache.pinot.common.metadata.segment.RealtimeSegmentZKMetadata;
import org.apache.pinot.common.metadata.segment.SegmentPartitionMetadata;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
import org.apache.pinot.common.metrics.ControllerMeter;
@@ -161,82 +163,84 @@ public class PinotLLCRealtimeSegmentManager {
_flushThresholdUpdateManager = new FlushThresholdUpdateManager();
}
+
/**
- * The committing segment will call this.
- *
- * For example, say we have 3 shards, grouped into PartitionGroups as [0], [1], [2]
- * Now segment of PG (partition group) 0 is committing. First, we'll update the metadata to DONE, and ideal state to ONLINE
- * Then, the currentPartitionGroupMetadata list will contain - [1], [2]
- * The newPartitionGroupMetadata list will contain - [0], [1], [2]
- * We then get the set of PGs for which new segments need to be made - [0]
+ * Using the ideal state and segment metadata, return a list of the current partition groups
*/
- public void commitPartitionGroup(String realtimeTableName, CommittingSegmentDescriptor committingSegmentDescriptor) {
- TableConfig realtimeTableConfig = getTableConfig(realtimeTableName);
- StreamConfig streamConfig = new StreamConfig(realtimeTableName, IngestionConfigUtils.getStreamConfigMap(realtimeTableConfig));
- int numReplicas = realtimeTableConfig.getValidationConfig().getReplicasPerPartitionNumber();
- IdealState idealState = getIdealState(realtimeTableName);
-
- // update status in segment metadata to DONE
- // ..
-
- // update Ideal State for this segment to ONLINE
- // ..
-
- // fetch current partition groups (which are actively CONSUMING - from example above, [1], [2])
- List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
-
- // get new partition groups (honor any groupings which are already consuming - [0], [1], [2])
- StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
- StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
- .createStreamMetadataProvider(streamConfig.getTopicName() + " " + System.currentTimeMillis());
- List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
-
- // from the above list, remove the partition groups which are already CONSUMING
- // i.e. newPartitionGroups - currentPartitionGroups. Therefore, ([0], [1], [2]) - ([1], [2]) = ([0])
- // ..
-
- // setup segment metadata and ideal state for the new found partition groups
- setupNewPartitionGroups(newPartitionGroupMetadataList, numReplicas);
- }
+ public List<PartitionGroupMetadata> getCurrentPartitionGroupMetadataList(IdealState idealState) {
+ List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- public void setupIdealStateForConsuming(List<SegmentZKMetadata> segmentZKMetadata, int numReplicas) {
- // add all segments from the list to ideal state, with state CONSUMING
- }
+ // from all segment names in the ideal state, find unique groups
+ Map<Integer, LLCSegmentName> groupIdToLatestSegment = new HashMap<>();
+ for (String segment : idealState.getPartitionSet()) {
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segment);
+ int partitionGroupId = llcSegmentName.getPartitionGroupId();
+ groupIdToLatestSegment.compute(partitionGroupId, (k, latestSegment) -> {
+ if (latestSegment == null) {
+ return llcSegmentName;
+ } else {
+ return latestSegment.getSequenceNumber() > llcSegmentName.getSequenceNumber() ? latestSegment
+ : llcSegmentName;
+ }
+ });
+ }
- public void persistSegmentMetadata(List<SegmentZKMetadata> segmentMetadata) {
- // persist new segment metadata from list to zk
+ // create a PartitionGroupMetadata for each latest segment
+ for (Map.Entry<Integer, LLCSegmentName> entry : groupIdToLatestSegment.entrySet()) {
+ int partitionGroupId = entry.getKey();
+ LLCSegmentName llcSegmentName = entry.getValue();
+ RealtimeSegmentZKMetadata realtimeSegmentZKMetadata = ZKMetadataProvider
+ .getRealtimeSegmentZKMetadata(_propertyStore, llcSegmentName.getTableName(), llcSegmentName.getSegmentName());
+ Preconditions.checkNotNull(realtimeSegmentZKMetadata);
+ LLCRealtimeSegmentZKMetadata llRealtimeSegmentZKMetadata =
+ (LLCRealtimeSegmentZKMetadata) realtimeSegmentZKMetadata;
+ PartitionGroupMetadata partitionGroupMetadata =
+ new PartitionGroupMetadata(partitionGroupId, llcSegmentName.getSequenceNumber(),
+ llRealtimeSegmentZKMetadata.getStartOffset(), llRealtimeSegmentZKMetadata.getEndOffset(),
+ llRealtimeSegmentZKMetadata.getStatus().toString());
+ partitionGroupMetadataList.add(partitionGroupMetadata);
+ }
+ return partitionGroupMetadataList;
}
/**
- * Using the list of partition group metadata, create a list of equivalent segment zk metadata
+ * Sets up the realtime table ideal state for a table of consumer type SHARDED
*/
- public List<SegmentZKMetadata> constructSegmentMetadata(List<PartitionGroupMetadata> partitionGroupMetadataList) {
- List<SegmentZKMetadata> segmentZKMetadata = new ArrayList<>();
- // for each partition group construct a segment zk metadata object
- return segmentZKMetadata;
- }
+ public void setupNewShardedTable(TableConfig tableConfig, IdealState idealState) {
+ Preconditions.checkState(!_isStopping, "Segment manager is stopping");
- /**
- * Using the ideal state, return a list of the current partition groups
- */
- public List<PartitionGroupMetadata> getCurrentPartitionGroupMetadataList(IdealState idealState) {
- List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
- // from all segment names in the ideal state, find unique groups
+ String realtimeTableName = tableConfig.getTableName();
+ LOGGER.info("Setting up new SHARDED table: {}", realtimeTableName);
- // create a PartitionGroupMetadata, one for each group
- return partitionGroupMetadataList;
- }
+ _flushThresholdUpdateManager.clearFlushThresholdUpdater(realtimeTableName);
- public void setupNewPartitionGroups(List<PartitionGroupMetadata> newPartitionGroupMetadataList, int numReplicas) {
- // construct segment zk metadata for the new partition groups
- List<SegmentZKMetadata> segmentMetadata = constructSegmentMetadata(newPartitionGroupMetadataList);
+ PartitionLevelStreamConfig streamConfig =
+ new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
+
+ // get new partition groups and their metadata
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
+ .createStreamMetadataProvider(streamConfig.getTopicName() + "_" + System.currentTimeMillis());
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList =
+ streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000);
+ int numPartitionGroups = newPartitionGroupMetadataList.size();
+
+ InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
+ int numReplicas = getNumReplicas(tableConfig, instancePartitions);
- // create these new segments metadata
- persistSegmentMetadata(segmentMetadata);
+ SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig);
+ Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
+ Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions);
- // setup ideal state for the new segments
- setupIdealStateForConsuming(segmentMetadata, numReplicas);
+ long currentTimeMs = getCurrentTimeMs();
+ Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
+ for (PartitionGroupMetadata partitionGroupMetadata : newPartitionGroupMetadataList) {
+ String segmentName = setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupMetadata.getPartitionGroupId(),
+ currentTimeMs, instancePartitions, numPartitionGroups, numReplicas);
+ updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment,
+ instancePartitionsMap);
+ }
+ setIdealState(realtimeTableName, idealState);
}
public boolean getIsSplitCommitEnabled() {
@@ -532,13 +536,50 @@ public class PinotLLCRealtimeSegmentManager {
_helixResourceManager.sendSegmentRefreshMessage(realtimeTableName, committingSegmentName, false, true);
// Step-2
+
+ // Say we currently were consuming from 3 shards A, B, C. Of those, A is the one committing. Also suppose that new partition D has come up
+ // get current partition groups - this gives current state of latest segments for each partition [A - DONE], [B - IN_PROGRESS], [C - IN_PROGRESS]
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
+ StreamConfig streamConfig = new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
+ .createStreamMetadataProvider(streamConfig.getTopicName() + " " + System.currentTimeMillis());
+ // find new partition groups [A],[B],[C],[D]
+ List<PartitionGroupMetadata> newPartitionGroupMetadataList =
+ streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
+
+ // create new segment metadata, only if it is not IN_PROGRESS in the current state
+ Map<Integer, PartitionGroupMetadata> currentGroupIdToMetadata = currentPartitionGroupMetadataList.stream().collect(
+ Collectors.toMap(PartitionGroupMetadata::getPartitionGroupId, p -> p));
+
+ List<String> newConsumingSegmentNames = new ArrayList<>();
+ String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
long newSegmentCreationTimeMs = getCurrentTimeMs();
- LLCSegmentName newLLCSegmentName =
- getNextLLCSegmentName(new LLCSegmentName(committingSegmentName), newSegmentCreationTimeMs);
- createNewSegmentZKMetadata(tableConfig,
- new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig)),
- newLLCSegmentName, newSegmentCreationTimeMs, committingSegmentDescriptor, committingSegmentZKMetadata,
- instancePartitions, numPartitions, numReplicas);
+ for (PartitionGroupMetadata partitionGroupMetadata : newPartitionGroupMetadataList) {
+ int newPartitionGroupId = partitionGroupMetadata.getPartitionGroupId();
+ PartitionGroupMetadata currentPartitionGroupMetadata = currentGroupIdToMetadata.get(newPartitionGroupId);
+ if (currentPartitionGroupMetadata == null) { // not present in current state
+ // make new segment
+ LLCSegmentName newLLCSegmentName =
+ new LLCSegmentName(rawTableName, newPartitionGroupId, STARTING_SEQUENCE_NUMBER, newSegmentCreationTimeMs);
+ createNewSegmentZKMetadata(tableConfig, new PartitionLevelStreamConfig(tableConfig.getTableName(),
+ IngestionConfigUtils.getStreamConfigMap(tableConfig)), newLLCSegmentName, newSegmentCreationTimeMs,
+ committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, numPartitions, numReplicas);
+ newConsumingSegmentNames.add(newLLCSegmentName.getSegmentName());
+ } else {
+ String currentStatus = currentPartitionGroupMetadata.getStatus();
+ if (!currentStatus.equals(Status.IN_PROGRESS.toString())) { // not IN_PROGRESS anymore in current state
+ // make new segment
+ LLCSegmentName newLLCSegmentName = new LLCSegmentName(rawTableName, newPartitionGroupId,
+ currentPartitionGroupMetadata.getSequenceNumber() + 1, newSegmentCreationTimeMs);
+ createNewSegmentZKMetadata(tableConfig, new PartitionLevelStreamConfig(tableConfig.getTableName(),
+ IngestionConfigUtils.getStreamConfigMap(tableConfig)), newLLCSegmentName, newSegmentCreationTimeMs,
+ committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, numPartitions, numReplicas);
+ newConsumingSegmentNames.add(newLLCSegmentName.getSegmentName());
+ }
+ }
+ }
+
// Step-3
SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig);
@@ -554,7 +595,7 @@ public class PinotLLCRealtimeSegmentManager {
Lock lock = _idealStateUpdateLocks[lockIndex];
try {
lock.lock();
- updateIdealStateOnSegmentCompletion(realtimeTableName, committingSegmentName, newLLCSegmentName.getSegmentName(),
+ updateIdealStateOnSegmentCompletion(realtimeTableName, committingSegmentName, newConsumingSegmentNames,
segmentAssignment, instancePartitionsMap);
} finally {
lock.unlock();
@@ -845,7 +886,7 @@ public class PinotLLCRealtimeSegmentManager {
*/
@VisibleForTesting
void updateIdealStateOnSegmentCompletion(String realtimeTableName, String committingSegmentName,
- String newSegmentName, SegmentAssignment segmentAssignment,
+ List<String> newSegmentNames, SegmentAssignment segmentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
assert idealState != null;
@@ -862,14 +903,18 @@ public class PinotLLCRealtimeSegmentManager {
"Exceeded max segment completion time for segment " + committingSegmentName);
}
updateInstanceStatesForNewConsumingSegment(idealState.getRecord().getMapFields(), committingSegmentName,
- newSegmentName, segmentAssignment, instancePartitionsMap);
+ null, segmentAssignment, instancePartitionsMap);
+ for (String newSegmentName : newSegmentNames) {
+ updateInstanceStatesForNewConsumingSegment(idealState.getRecord().getMapFields(), null,
+ newSegmentName, segmentAssignment, instancePartitionsMap);
+ }
return idealState;
}, RetryPolicies.exponentialBackoffRetryPolicy(10, 1000L, 1.2f));
}
@VisibleForTesting
void updateInstanceStatesForNewConsumingSegment(Map<String, Map<String, String>> instanceStatesMap,
- @Nullable String committingSegmentName, String newSegmentName, SegmentAssignment segmentAssignment,
+ @Nullable String committingSegmentName, @Nullable String newSegmentName, SegmentAssignment segmentAssignment,
Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap) {
if (committingSegmentName != null) {
// Change committing segment state to ONLINE
@@ -880,11 +925,11 @@ public class PinotLLCRealtimeSegmentManager {
}
// Assign instances to the new segment and add instances as state CONSUMING
- List<String> instancesAssigned =
- segmentAssignment.assignSegment(newSegmentName, instanceStatesMap, instancePartitionsMap);
- instanceStatesMap.put(newSegmentName,
- SegmentAssignmentUtils.getInstanceStateMap(instancesAssigned, SegmentStateModel.CONSUMING));
- LOGGER.info("Adding new CONSUMING segment: {} to instances: {}", newSegmentName, instancesAssigned);
+ if (newSegmentName != null) {
+ List<String> instancesAssigned = segmentAssignment.assignSegment(newSegmentName, instanceStatesMap, instancePartitionsMap);
+ instanceStatesMap.put(newSegmentName, SegmentAssignmentUtils.getInstanceStateMap(instancesAssigned, SegmentStateModel.CONSUMING));
+ LOGGER.info("Adding new CONSUMING segment: {} to instances: {}", newSegmentName, instancesAssigned);
+ }
}
/*
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java
deleted file mode 100644
index 78ee12c..0000000
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java
+++ /dev/null
@@ -1,48 +0,0 @@
-package org.apache.pinot.core.realtime.impl.fakestream;
-
-import org.apache.pinot.spi.stream.Checkpoint;
-import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-
-
-public class FakePartitionGroupMetadata implements PartitionGroupMetadata {
-
- private final int _groupId;
- public FakePartitionGroupMetadata(int groupId) {
- _groupId = groupId;
- }
-
- @Override
- public int getGroupId() {
- return getGroupId();
- }
-
- @Override
- public Checkpoint getStartCheckpoint() {
- return null;
- }
-
- @Override
- public Checkpoint getEndCheckpoint() {
- return null;
- }
-
- @Override
- public void setStartCheckpoint(Checkpoint startCheckpoint) {
-
- }
-
- @Override
- public void setEndCheckpoint(Checkpoint endCheckpoint) {
-
- }
-
- @Override
- public byte[] serialize() {
- return new byte[0];
- }
-
- @Override
- public PartitionGroupMetadata deserialize(byte[] blob) {
- return null;
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java
deleted file mode 100644
index 1d792ac..0000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java
+++ /dev/null
@@ -1,48 +0,0 @@
-package org.apache.pinot.plugin.stream.kafka09;
-
-import org.apache.pinot.spi.stream.Checkpoint;
-import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-
-
-public class KafkaPartitionGroupMetadata implements PartitionGroupMetadata {
-
- private final int _groupId;
- public KafkaPartitionGroupMetadata(int partitionId) {
- _groupId = partitionId;
- }
-
- @Override
- public int getGroupId() {
- return _groupId;
- }
-
- @Override
- public Checkpoint getStartCheckpoint() {
- return null;
- }
-
- @Override
- public Checkpoint getEndCheckpoint() {
- return null;
- }
-
- @Override
- public void setStartCheckpoint(Checkpoint startCheckpoint) {
-
- }
-
- @Override
- public void setEndCheckpoint(Checkpoint endCheckpoint) {
-
- }
-
- @Override
- public byte[] serialize() {
- return new byte[0];
- }
-
- @Override
- public PartitionGroupMetadata deserialize(byte[] blob) {
- return null;
- }
-}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java
deleted file mode 100644
index 31ae75a..0000000
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java
+++ /dev/null
@@ -1,48 +0,0 @@
-package org.apache.pinot.plugin.stream.kafka20;
-
-import org.apache.pinot.spi.stream.Checkpoint;
-import org.apache.pinot.spi.stream.PartitionGroupMetadata;
-
-
-public class KafkaPartitionGroupMetadata implements PartitionGroupMetadata {
-
- private final int _groupId;
- public KafkaPartitionGroupMetadata(int partitionId) {
- _groupId = partitionId;
- }
-
- @Override
- public int getGroupId() {
- return _groupId;
- }
-
- @Override
- public Checkpoint getStartCheckpoint() {
- return null;
- }
-
- @Override
- public Checkpoint getEndCheckpoint() {
- return null;
- }
-
- @Override
- public void setStartCheckpoint(Checkpoint startCheckpoint) {
-
- }
-
- @Override
- public void setEndCheckpoint(Checkpoint endCheckpoint) {
-
- }
-
- @Override
- public byte[] serialize() {
- return new byte[0];
- }
-
- @Override
- public PartitionGroupMetadata deserialize(byte[] blob) {
- return null;
- }
-}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
index 0f44173..f662d99 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
@@ -18,22 +18,56 @@
*/
package org.apache.pinot.spi.stream;
-import java.util.List;
+public class PartitionGroupMetadata {
+ // fixme: Make partitionGroupId string everywhere (LLCSegmentName, StreamMetadataProvider)
+ private final int _partitionGroupId;
+ private int _sequenceNumber;
+ private String _startCheckpoint;
+ private String _endCheckpoint;
+ private String _status;
-public interface PartitionGroupMetadata {
+ public PartitionGroupMetadata(int partitionGroupId, int sequenceNumber, String startCheckpoint,
+ String endCheckpoint, String status) {
+ _partitionGroupId = partitionGroupId;
+ _sequenceNumber = sequenceNumber;
+ _startCheckpoint = startCheckpoint;
+ _endCheckpoint = endCheckpoint;
+ }
- int getGroupId();
+ public void setSequenceNumber(int sequenceNumber) {
+ _sequenceNumber = sequenceNumber;
+ }
- Checkpoint getStartCheckpoint(); // similar to getStartOffset
+ public void setStartCheckpoint(String startCheckpoint) {
+ _startCheckpoint = startCheckpoint;
+ }
- Checkpoint getEndCheckpoint(); // similar to getEndOffset
+ public void setEndCheckpoint(String endCheckpoint) {
+ _endCheckpoint = endCheckpoint;
+ }
- void setStartCheckpoint(Checkpoint startCheckpoint);
+ public int getPartitionGroupId() {
+ return _partitionGroupId;
+ }
- void setEndCheckpoint(Checkpoint endCheckpoint);
+ public int getSequenceNumber() {
+ return _sequenceNumber;
+ }
- byte[] serialize();
+ public String getStartCheckpoint() {
+ return _startCheckpoint;
+ }
- PartitionGroupMetadata deserialize(byte[] blob);
+ public String getEndCheckpoint() {
+ return _endCheckpoint;
+ }
+
+ public String getStatus() {
+ return _status;
+ }
+
+ public void setStatus(String status) {
+ _status = status;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 02/08: StreamPartitionOffset to implement
Checkpoint
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 652a3e68462fe007b59a3505c44bba3f0f190c3b
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Tue Dec 29 18:43:48 2020 -0800
StreamPartitionOffset to implement Checkpoint
---
.../common/metadata/segment/RealtimeSegmentZKMetadata.java | 6 ++++++
.../org/apache/pinot/common/utils/CommonConstants.java | 4 ++++
.../impl/fakestream/FakeStreamConsumerFactory.java | 14 ++++++++++++++
.../tests/FlakyConsumerRealtimeClusterIntegrationTest.java | 14 ++++++++++++++
.../pinot/plugin/stream/kafka09/KafkaConsumerFactory.java | 14 ++++++++++++++
.../pinot/plugin/stream/kafka20/KafkaConsumerFactory.java | 14 ++++++++++++++
.../main/java/org/apache/pinot/spi/stream/Checkpoint.java | 6 +++---
.../java/org/apache/pinot/spi/stream/LongMsgOffset.java | 10 ++++++++++
.../apache/pinot/spi/stream/PartitionGroupConsumer.java | 5 ++++-
.../apache/pinot/spi/stream/PartitionGroupMetadata.java | 2 --
.../apache/pinot/spi/stream/StreamPartitionMsgOffset.java | 2 +-
11 files changed, 84 insertions(+), 7 deletions(-)
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java b/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
index d88be18..c46af53 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metadata/segment/RealtimeSegmentZKMetadata.java
@@ -35,6 +35,7 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
private Status _status = null;
private int _sizeThresholdToFlushSegment = -1;
private String _timeThresholdToFlushSegment = null; // store as period string for readability
+ private String _partitionGroupMetadataStr = null;
public RealtimeSegmentZKMetadata() {
setSegmentType(SegmentType.REALTIME);
@@ -49,6 +50,7 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
if (flushThresholdTime != null && !flushThresholdTime.equals(NULL)) {
_timeThresholdToFlushSegment = znRecord.getSimpleField(CommonConstants.Segment.FLUSH_THRESHOLD_TIME);
}
+ _partitionGroupMetadataStr = znRecord.getSimpleField(CommonConstants.Segment.PARTITION_GROUP_METADATA);
}
@Override
@@ -141,4 +143,8 @@ public class RealtimeSegmentZKMetadata extends SegmentZKMetadata {
public void setTimeThresholdToFlushSegment(String timeThresholdPeriodString) {
_timeThresholdToFlushSegment = timeThresholdPeriodString;
}
+
+ public String getPartitionGroupMetadataStr() {
+ return _partitionGroupMetadataStr;
+ }
}
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
index 9773e7e..7a91d8c 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/CommonConstants.java
@@ -379,6 +379,10 @@ public class CommonConstants {
public static final String FLUSH_THRESHOLD_TIME = "segment.flush.threshold.time";
public static final String PARTITION_METADATA = "segment.partition.metadata";
/**
+ * Serialized {@link org.apache.pinot.spi.stream.PartitionGroupMetadata} for this segment
+ */
+ public static final String PARTITION_GROUP_METADATA = "segment.partition.group.metadata";
+ /**
* This field is used for parallel push protection to lock the segment globally.
* We put the segment upload start timestamp so that if the previous push failed without unlock the segment, the
* next upload won't be blocked forever.
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
index bb01e5c..9669223 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.realtime.impl.fakestream;
+import java.util.List;
import java.util.Set;
import org.apache.pinot.core.util.IngestionUtils;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -26,6 +27,8 @@ import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.MessageBatch;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
@@ -66,6 +69,17 @@ public class FakeStreamConsumerFactory extends StreamConsumerFactory {
return new FakeStreamMetadataProvider(_streamConfig);
}
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
+ return null;
+ }
+
+ @Override
+ public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ return null;
+ }
+
public static void main(String[] args)
throws Exception {
String clientId = "client_id_localhost_tester";
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
index b05244f..808a464 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
@@ -19,9 +19,12 @@
package org.apache.pinot.integration.tests;
import java.lang.reflect.Constructor;
+import java.util.List;
import java.util.Random;
import java.util.Set;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
@@ -117,5 +120,16 @@ public class FlakyConsumerRealtimeClusterIntegrationTest extends RealtimeCluster
public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
+ return null;
+ }
+
+ @Override
+ public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ return null;
+ }
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
index 615e354..b8ed19d 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
@@ -18,7 +18,10 @@
*/
package org.apache.pinot.plugin.stream.kafka09;
+import java.util.List;
import java.util.Set;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamLevelConsumer;
@@ -50,4 +53,15 @@ public class KafkaConsumerFactory extends StreamConsumerFactory {
public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
return new KafkaStreamMetadataProvider(clientId, _streamConfig);
}
+
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
+ return null;
+ }
+
+ @Override
+ public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ return null;
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
index e0d1015..806baff 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
@@ -18,7 +18,10 @@
*/
package org.apache.pinot.plugin.stream.kafka20;
+import java.util.List;
import java.util.Set;
+import org.apache.pinot.spi.stream.PartitionGroupConsumer;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelConsumer;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
import org.apache.pinot.spi.stream.StreamLevelConsumer;
@@ -47,4 +50,15 @@ public class KafkaConsumerFactory extends StreamConsumerFactory {
public StreamMetadataProvider createStreamMetadataProvider(String clientId) {
return new KafkaStreamMetadataProvider(clientId, _streamConfig);
}
+
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
+ return null;
+ }
+
+ @Override
+ public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ return null;
+ }
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java
index 627c964..bae8832 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/Checkpoint.java
@@ -18,7 +18,7 @@
*/
package org.apache.pinot.spi.stream;
-public interface Checkpoint {
- byte[] serialize();
- Checkpoint deserialize(byte[] blob);
+public interface Checkpoint extends Comparable {
+ String serialize();
+ Checkpoint deserialize(String checkpointStr);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
index e5025f6..e8fa275 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/LongMsgOffset.java
@@ -50,4 +50,14 @@ public class LongMsgOffset implements StreamPartitionMsgOffset {
public String toString() {
return Long.toString(_offset);
}
+
+ @Override
+ public String serialize() {
+ return toString();
+ }
+
+ @Override
+ public Checkpoint deserialize(String checkpointStr) {
+ return new LongMsgOffset(checkpointStr);
+ }
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
index 2f138c2..e096e67 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupConsumer.java
@@ -18,6 +18,9 @@
*/
package org.apache.pinot.spi.stream;
-public interface PartitionGroupConsumer {
+import java.io.Closeable;
+
+
+public interface PartitionGroupConsumer extends Closeable {
FetchResult fetch(Checkpoint start, Checkpoint end, long timeout);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
index 779c167..0f44173 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadata.java
@@ -25,8 +25,6 @@ public interface PartitionGroupMetadata {
int getGroupId();
- List<String> getPartitions();
-
Checkpoint getStartCheckpoint(); // similar to getStartOffset
Checkpoint getEndCheckpoint(); // similar to getEndOffset
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
index 72654bf..06a090e 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamPartitionMsgOffset.java
@@ -39,7 +39,7 @@ import org.apache.pinot.spi.annotations.InterfaceStability;
* versions of the stream implementation
*/
@InterfaceStability.Evolving
-public interface StreamPartitionMsgOffset extends Comparable {
+public interface StreamPartitionMsgOffset extends Checkpoint {
/**
* Compare this offset with another one.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 03/08: Rename partitionId to partitionGroupId
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 3ef1870132bfd5867fd028ce958b53f7423afb90
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Wed Dec 30 13:46:22 2020 -0800
Rename partitionId to partitionGroupId
---
.../segmentselector/RealtimeSegmentSelector.java | 2 +-
.../apache/pinot/common/utils/LLCSegmentName.java | 24 +++++-----
.../org/apache/pinot/common/utils/SegmentName.java | 2 +-
.../pinot/common/utils/SegmentNameBuilderTest.java | 6 +--
.../helix/core/PinotHelixResourceManager.java | 5 ++-
.../helix/core/PinotTableIdealStateBuilder.java | 15 ++++---
.../segment/RealtimeSegmentAssignment.java | 6 +--
.../RealtimeToOfflineSegmentsTaskGenerator.java | 4 +-
.../realtime/PinotLLCRealtimeSegmentManager.java | 51 ++++++++++++----------
.../SegmentSizeBasedFlushThresholdUpdater.java | 2 +-
.../PinotLLCRealtimeSegmentManagerTest.java | 16 ++++---
.../realtime/LLRealtimeSegmentDataManager.java | 34 ++++++++-------
.../manager/realtime/RealtimeTableDataManager.java | 13 +++---
.../realtime/LLRealtimeSegmentDataManagerTest.java | 10 ++---
.../fakestream/FakePartitionGroupMetadata.java | 48 ++++++++++++++++++++
.../impl/fakestream/FakeStreamConsumerFactory.java | 10 +----
.../fakestream/FakeStreamMetadataProvider.java | 15 ++++++-
...lakyConsumerRealtimeClusterIntegrationTest.java | 9 +---
...PartitionLLCRealtimeClusterIntegrationTest.java | 6 +--
.../stream/kafka09/KafkaConsumerFactory.java | 9 +---
.../kafka09/KafkaPartitionGroupMetadata.java | 48 ++++++++++++++++++++
.../kafka09/KafkaStreamMetadataProvider.java | 26 +++++++++++
.../kafka09/KafkaPartitionLevelConsumerTest.java | 2 +-
.../stream/kafka20/KafkaConsumerFactory.java | 9 +---
.../kafka20/KafkaPartitionGroupMetadata.java | 48 ++++++++++++++++++++
.../kafka20/KafkaStreamMetadataProvider.java | 21 +++++++++
...her.java => PartitionGroupMetadataFetcher.java} | 18 +++++---
.../pinot/spi/stream/PartitionOffsetFetcher.java | 15 ++++---
.../pinot/spi/stream/StreamConsumerFactory.java | 8 +---
.../pinot/spi/stream/StreamMetadataProvider.java | 9 +++-
30 files changed, 347 insertions(+), 144 deletions(-)
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentselector/RealtimeSegmentSelector.java b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentselector/RealtimeSegmentSelector.java
index f462326..2d778c6 100644
--- a/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentselector/RealtimeSegmentSelector.java
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/routing/segmentselector/RealtimeSegmentSelector.java
@@ -95,7 +95,7 @@ public class RealtimeSegmentSelector implements SegmentSelector {
if (instanceStateMap.containsValue(SegmentStateModel.CONSUMING)) {
// Keep the first CONSUMING segment for each partition
LLCSegmentName llcSegmentName = new LLCSegmentName(segment);
- partitionIdToFirstConsumingLLCSegmentMap.compute(llcSegmentName.getPartitionId(), (k, consumingSegment) -> {
+ partitionIdToFirstConsumingLLCSegmentMap.compute(llcSegmentName.getPartitionGroupId(), (k, consumingSegment) -> {
if (consumingSegment == null) {
return llcSegmentName;
} else {
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 adc24ad..a66bb3c 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
@@ -26,7 +26,7 @@ import org.joda.time.DateTimeZone;
public class LLCSegmentName extends SegmentName implements Comparable {
private final static String DATE_FORMAT = "yyyyMMdd'T'HHmm'Z'";
private final String _tableName;
- private final int _partitionId;
+ private final int _partitionGroupId;
private final int _sequenceNumber;
private final String _creationTime;
private final String _segmentName;
@@ -39,22 +39,22 @@ public class LLCSegmentName extends SegmentName implements Comparable {
_segmentName = segmentName;
String[] parts = StringUtils.splitByWholeSeparator(segmentName, SEPARATOR);
_tableName = parts[0];
- _partitionId = Integer.parseInt(parts[1]);
+ _partitionGroupId = Integer.parseInt(parts[1]);
_sequenceNumber = Integer.parseInt(parts[2]);
_creationTime = parts[3];
}
- public LLCSegmentName(String tableName, int partitionId, int sequenceNumber, long msSinceEpoch) {
+ public LLCSegmentName(String tableName, int partitionGroupId, int sequenceNumber, long msSinceEpoch) {
if (!isValidComponentName(tableName)) {
throw new RuntimeException("Invalid table name " + tableName);
}
_tableName = tableName;
- _partitionId = partitionId;
+ _partitionGroupId = partitionGroupId;
_sequenceNumber = sequenceNumber;
// ISO8601 date: 20160120T1234Z
DateTime dateTime = new DateTime(msSinceEpoch, DateTimeZone.UTC);
_creationTime = dateTime.toString(DATE_FORMAT);
- _segmentName = tableName + SEPARATOR + partitionId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
+ _segmentName = tableName + SEPARATOR + partitionGroupId + SEPARATOR + sequenceNumber + SEPARATOR + _creationTime;
}
/**
@@ -75,13 +75,13 @@ public class LLCSegmentName extends SegmentName implements Comparable {
}
@Override
- public int getPartitionId() {
- return _partitionId;
+ public int getPartitionGroupId() {
+ return _partitionGroupId;
}
@Override
public String getPartitionRange() {
- return Integer.toString(getPartitionId());
+ return Integer.toString(getPartitionGroupId());
}
@Override
@@ -110,9 +110,9 @@ public class LLCSegmentName extends SegmentName implements Comparable {
throw new RuntimeException(
"Cannot compare segment names " + this.getSegmentName() + " and " + other.getSegmentName());
}
- if (this.getPartitionId() > other.getPartitionId()) {
+ if (this.getPartitionGroupId() > other.getPartitionGroupId()) {
return 1;
- } else if (this.getPartitionId() < other.getPartitionId()) {
+ } else if (this.getPartitionGroupId() < other.getPartitionGroupId()) {
return -1;
} else {
if (this.getSequenceNumber() > other.getSequenceNumber()) {
@@ -141,7 +141,7 @@ public class LLCSegmentName extends SegmentName implements Comparable {
LLCSegmentName segName = (LLCSegmentName) o;
- if (_partitionId != segName._partitionId) {
+ if (_partitionGroupId != segName._partitionGroupId) {
return false;
}
if (_sequenceNumber != segName._sequenceNumber) {
@@ -159,7 +159,7 @@ public class LLCSegmentName extends SegmentName implements Comparable {
@Override
public int hashCode() {
int result = _tableName != null ? _tableName.hashCode() : 0;
- result = 31 * result + _partitionId;
+ result = 31 * result + _partitionGroupId;
result = 31 * result + _sequenceNumber;
result = 31 * result + (_creationTime != null ? _creationTime.hashCode() : 0);
result = 31 * result + (_segmentName != null ? _segmentName.hashCode() : 0);
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 6763f6d..b0c00ae 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
@@ -63,7 +63,7 @@ public abstract class SegmentName {
throw new RuntimeException("No groupId in " + getSegmentName());
}
- public int getPartitionId() {
+ public int getPartitionGroupId() {
throw new RuntimeException("No partitionId in " + getSegmentName());
}
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
index f632f51..de606cc 100644
--- 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
@@ -58,7 +58,7 @@ public class SegmentNameBuilderTest {
// Check partition range
assertEquals(longNameSegment.getPartitionRange(), "0");
assertEquals(shortNameSegment.getPartitionRange(), "ALL");
- assertEquals(llcSegment.getPartitionId(), 0);
+ assertEquals(llcSegment.getPartitionGroupId(), 0);
// Check groupId
assertEquals(longNameSegment.getGroupId(), "myTable_REALTIME_1234567_0");
@@ -127,14 +127,14 @@ public class SegmentNameBuilderTest {
LLCSegmentName segName1 = new LLCSegmentName(tableName, partitionId, sequenceNumber, msSinceEpoch);
Assert.assertEquals(segName1.getSegmentName(), segmentName);
- Assert.assertEquals(segName1.getPartitionId(), partitionId);
+ Assert.assertEquals(segName1.getPartitionGroupId(), partitionId);
Assert.assertEquals(segName1.getCreationTime(), creationTime);
Assert.assertEquals(segName1.getSequenceNumber(), sequenceNumber);
Assert.assertEquals(segName1.getTableName(), tableName);
LLCSegmentName segName2 = new LLCSegmentName(segmentName);
Assert.assertEquals(segName2.getSegmentName(), segmentName);
- Assert.assertEquals(segName2.getPartitionId(), partitionId);
+ Assert.assertEquals(segName2.getPartitionGroupId(), partitionId);
Assert.assertEquals(segName2.getCreationTime(), creationTime);
Assert.assertEquals(segName2.getSequenceNumber(), sequenceNumber);
Assert.assertEquals(segName2.getTableName(), tableName);
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 5afbe7e..7f36e5a 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
@@ -127,6 +127,7 @@ import org.apache.pinot.spi.stream.PartitionGroupMetadata;
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.StreamMetadataProvider;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -1378,6 +1379,8 @@ public class PinotHelixResourceManager {
*/
private void setupShardedRealtimeTable(StreamConfig streamConfig, IdealState idealState, int numReplicas) {
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
+ .createStreamMetadataProvider(streamConfig.getTopicName() + "_" + System.currentTimeMillis());
// get current partition groups and their metadata - this will be empty when creating the table
List<PartitionGroupMetadata> currentPartitionGroupMetadataList = _pinotLLCRealtimeSegmentManager.getCurrentPartitionGroupMetadataList(idealState);
@@ -1385,7 +1388,7 @@ public class PinotHelixResourceManager {
// get new partition groups and their metadata,
// Assume table has 3 shards. Say we get [0], [1], [2] groups (for now assume that each group contains only 1 shard)
List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamConsumerFactory.getPartitionGroupMetadataList(currentPartitionGroupMetadataList);
+ streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 5000);
// setup segment zk metadata and ideal state for all the new found partition groups
_pinotLLCRealtimeSegmentManager.setupNewPartitionGroups(newPartitionGroupMetadataList, numReplicas);
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 a564542..1e95966 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
@@ -32,7 +32,8 @@ import org.apache.pinot.common.utils.config.TagNameUtils;
import org.apache.pinot.common.utils.helix.HelixHelper;
import org.apache.pinot.controller.helix.core.realtime.PinotLLCRealtimeSegmentManager;
import org.apache.pinot.spi.config.table.TableConfig;
-import org.apache.pinot.spi.stream.PartitionCountFetcher;
+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.retry.RetryPolicies;
@@ -115,13 +116,15 @@ public class PinotTableIdealStateBuilder {
pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
}
- public static int getPartitionCount(StreamConfig streamConfig) {
- PartitionCountFetcher partitionCountFetcher = new PartitionCountFetcher(streamConfig);
+ public static List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig,
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ PartitionGroupMetadataFetcher partitionGroupMetadataFetcher =
+ new PartitionGroupMetadataFetcher(streamConfig, currentPartitionGroupMetadataList);
try {
- RetryPolicies.noDelayRetryPolicy(3).attempt(partitionCountFetcher);
- return partitionCountFetcher.getPartitionCount();
+ RetryPolicies.noDelayRetryPolicy(3).attempt(partitionGroupMetadataFetcher);
+ return partitionGroupMetadataFetcher.getPartitionGroupMetadataList();
} catch (Exception e) {
- Exception fetcherException = partitionCountFetcher.getException();
+ Exception fetcherException = partitionGroupMetadataFetcher.getException();
LOGGER.error("Could not get partition count for {}", streamConfig.getTopicName(), fetcherException);
throw new RuntimeException(fetcherException);
}
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
index a069734..e27958f 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java
@@ -136,7 +136,7 @@ public class RealtimeSegmentAssignment implements SegmentAssignment {
* Helper method to assign instances for CONSUMING segment based on the segment partition id and instance partitions.
*/
private List<String> assignConsumingSegment(String segmentName, InstancePartitions instancePartitions) {
- int partitionId = new LLCSegmentName(segmentName).getPartitionId();
+ int partitionId = new LLCSegmentName(segmentName).getPartitionGroupId();
int numReplicaGroups = instancePartitions.getNumReplicaGroups();
if (numReplicaGroups == 1) {
@@ -325,7 +325,7 @@ public class RealtimeSegmentAssignment implements SegmentAssignment {
Map<Integer, List<String>> partitionIdToSegmentsMap = new HashMap<>();
for (String segmentName : currentAssignment.keySet()) {
- int partitionId = new LLCSegmentName(segmentName).getPartitionId();
+ int partitionId = new LLCSegmentName(segmentName).getPartitionGroupId();
partitionIdToSegmentsMap.computeIfAbsent(partitionId, k -> new ArrayList<>()).add(segmentName);
}
@@ -360,7 +360,7 @@ public class RealtimeSegmentAssignment implements SegmentAssignment {
// Replica-group based assignment
// Uniformly spray the segment partitions over the instance partitions
- int segmentPartitionId = new LLCSegmentName(segmentName).getPartitionId();
+ int segmentPartitionId = new LLCSegmentName(segmentName).getPartitionGroupId();
int numPartitions = instancePartitions.getNumPartitions();
int partitionId = segmentPartitionId % numPartitions;
return SegmentAssignmentUtils.assignSegmentWithReplicaGroup(currentAssignment, instancePartitions, partitionId);
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
index a0e519b..790522d 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/generator/RealtimeToOfflineSegmentsTaskGenerator.java
@@ -248,11 +248,11 @@ public class RealtimeToOfflineSegmentsTaskGenerator implements PinotTaskGenerato
Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
for (LLCRealtimeSegmentZKMetadata metadata : realtimeSegmentsMetadataList) {
LLCSegmentName llcSegmentName = new LLCSegmentName(metadata.getSegmentName());
- allPartitions.add(llcSegmentName.getPartitionId());
+ allPartitions.add(llcSegmentName.getPartitionGroupId());
if (metadata.getStatus().equals(Segment.Realtime.Status.DONE)) {
completedSegmentsMetadataList.add(metadata);
- latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionId(), (partitionId, latestLLCSegmentName) -> {
+ latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionGroupId(), (partitionId, latestLLCSegmentName) -> {
if (latestLLCSegmentName == null) {
return llcSegmentName;
} else {
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 b459760..6b220f2 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
@@ -83,6 +83,7 @@ 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.StreamConsumerFactoryProvider;
+import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffsetFactory;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
@@ -186,8 +187,10 @@ public class PinotLLCRealtimeSegmentManager {
// get new partition groups (honor any groupings which are already consuming - [0], [1], [2])
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
+ .createStreamMetadataProvider(streamConfig.getTopicName() + " " + System.currentTimeMillis());
List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamConsumerFactory.getPartitionGroupMetadataList(currentPartitionGroupMetadataList);
+ streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
// from the above list, remove the partition groups which are already CONSUMING
// i.e. newPartitionGroups - currentPartitionGroups. Therefore, ([0], [1], [2]) - ([1], [2]) = ([0])
@@ -292,7 +295,8 @@ public class PinotLLCRealtimeSegmentManager {
PartitionLevelStreamConfig streamConfig =
new PartitionLevelStreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
- int numPartitions = getNumPartitions(streamConfig);
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
+ int numPartitionGroups = getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList).size();
int numReplicas = getNumReplicas(tableConfig, instancePartitions);
SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig);
@@ -301,9 +305,9 @@ public class PinotLLCRealtimeSegmentManager {
long currentTimeMs = getCurrentTimeMs();
Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
- for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+ for (int partitionGroupId = 0; partitionGroupId < numPartitionGroups; partitionGroupId++) {
String segmentName =
- setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions,
+ setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupId, currentTimeMs, instancePartitions, numPartitionGroups,
numReplicas);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment,
instancePartitionsMap);
@@ -635,7 +639,7 @@ public class PinotLLCRealtimeSegmentManager {
// Add the partition metadata if available
SegmentPartitionMetadata partitionMetadata =
- getPartitionMetadataFromTableConfig(tableConfig, newLLCSegmentName.getPartitionId());
+ getPartitionMetadataFromTableConfig(tableConfig, newLLCSegmentName.getPartitionGroupId());
if (partitionMetadata != null) {
newSegmentZKMetadata.setPartitionMetadata(partitionMetadata);
}
@@ -705,22 +709,23 @@ public class PinotLLCRealtimeSegmentManager {
}
@VisibleForTesting
- int getNumPartitions(StreamConfig streamConfig) {
- return PinotTableIdealStateBuilder.getPartitionCount(streamConfig);
+ List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig,
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ return PinotTableIdealStateBuilder.getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList);
}
@VisibleForTesting
StreamPartitionMsgOffset getPartitionOffset(StreamConfig streamConfig, OffsetCriteria offsetCriteria,
- int partitionId) {
+ int partitionGroupId) {
PartitionOffsetFetcher partitionOffsetFetcher =
- new PartitionOffsetFetcher(offsetCriteria, partitionId, streamConfig);
+ new PartitionOffsetFetcher(offsetCriteria, partitionGroupId, streamConfig);
try {
RetryPolicies.fixedDelayRetryPolicy(3, 1000L).attempt(partitionOffsetFetcher);
return partitionOffsetFetcher.getOffset();
} catch (Exception e) {
throw new IllegalStateException(String
.format("Failed to fetch the offset for topic: %s, partition: %s with criteria: %s",
- streamConfig.getTopicName(), partitionId, offsetCriteria), e);
+ streamConfig.getTopicName(), partitionGroupId, offsetCriteria), e);
}
}
@@ -768,7 +773,7 @@ public class PinotLLCRealtimeSegmentManager {
Map<Integer, LLCSegmentName> latestLLCSegmentNameMap = new HashMap<>();
for (String segmentName : segments) {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionId(), (partitionId, latestLLCSegmentName) -> {
+ latestLLCSegmentNameMap.compute(llcSegmentName.getPartitionGroupId(), (partitionId, latestLLCSegmentName) -> {
if (latestLLCSegmentName == null) {
return llcSegmentName;
} else {
@@ -821,10 +826,12 @@ public class PinotLLCRealtimeSegmentManager {
Preconditions.checkState(!_isStopping, "Segment manager is stopping");
String realtimeTableName = tableConfig.getTableName();
- int numPartitions = getNumPartitions(streamConfig);
HelixHelper.updateIdealState(_helixManager, realtimeTableName, idealState -> {
assert idealState != null;
if (idealState.isEnabled()) {
+ List<PartitionGroupMetadata> currentPartitionGroupMetadataList =
+ getCurrentPartitionGroupMetadataList(idealState);
+ int numPartitions = getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList).size();
return ensureAllPartitionsConsuming(tableConfig, streamConfig, idealState, numPartitions);
} else {
LOGGER.info("Skipping LLC segments validation for disabled table: {}", realtimeTableName);
@@ -1066,7 +1073,7 @@ public class PinotLLCRealtimeSegmentManager {
String previousConsumingSegment = null;
for (Map.Entry<String, Map<String, String>> segmentEntry : instanceStatesMap.entrySet()) {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentEntry.getKey());
- if (llcSegmentName.getPartitionId() == partitionId && segmentEntry.getValue()
+ if (llcSegmentName.getPartitionGroupId() == partitionId && segmentEntry.getValue()
.containsValue(SegmentStateModel.CONSUMING)) {
previousConsumingSegment = llcSegmentName.getSegmentName();
break;
@@ -1091,7 +1098,7 @@ public class PinotLLCRealtimeSegmentManager {
for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
if (!latestSegmentZKMetadataMap.containsKey(partitionId)) {
String newSegmentName =
- setupNewPartition(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions,
+ setupNewPartitionGroup(tableConfig, streamConfig, partitionId, currentTimeMs, instancePartitions, numPartitions,
numReplicas);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, newSegmentName, segmentAssignment,
instancePartitionsMap);
@@ -1102,7 +1109,7 @@ public class PinotLLCRealtimeSegmentManager {
}
private LLCSegmentName getNextLLCSegmentName(LLCSegmentName lastLLCSegmentName, long creationTimeMs) {
- return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getPartitionId(),
+ return new LLCSegmentName(lastLLCSegmentName.getTableName(), lastLLCSegmentName.getPartitionGroupId(),
lastLLCSegmentName.getSequenceNumber() + 1, creationTimeMs);
}
@@ -1110,21 +1117,21 @@ public class PinotLLCRealtimeSegmentManager {
* Sets up a new partition.
* <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
*/
- private String setupNewPartition(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig, int partitionId,
- long creationTimeMs, InstancePartitions instancePartitions, int numPartitions, int numReplicas) {
+ private String setupNewPartitionGroup(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig, int partitionGroupId,
+ long creationTimeMs, InstancePartitions instancePartitions, int numPartitionGroups, int numReplicas) {
String realtimeTableName = tableConfig.getTableName();
- LOGGER.info("Setting up new partition: {} for table: {}", partitionId, realtimeTableName);
+ LOGGER.info("Setting up new partition group: {} for table: {}", partitionGroupId, realtimeTableName);
String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
LLCSegmentName newLLCSegmentName =
- new LLCSegmentName(rawTableName, partitionId, STARTING_SEQUENCE_NUMBER, creationTimeMs);
+ new LLCSegmentName(rawTableName, partitionGroupId, STARTING_SEQUENCE_NUMBER, creationTimeMs);
String newSegmentName = newLLCSegmentName.getSegmentName();
StreamPartitionMsgOffset startOffset =
- getPartitionOffset(streamConfig, streamConfig.getOffsetCriteria(), partitionId);
+ getPartitionOffset(streamConfig, streamConfig.getOffsetCriteria(), partitionGroupId);
CommittingSegmentDescriptor committingSegmentDescriptor =
new CommittingSegmentDescriptor(null, startOffset.toString(), 0);
createNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegmentName, creationTimeMs,
- committingSegmentDescriptor, null, instancePartitions, numPartitions, numReplicas);
+ committingSegmentDescriptor, null, instancePartitions, numPartitionGroups, numReplicas);
return newSegmentName;
}
@@ -1138,7 +1145,7 @@ public class PinotLLCRealtimeSegmentManager {
int numPartitions = 0;
for (String segmentName : idealState.getRecord().getMapFields().keySet()) {
if (LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)) {
- numPartitions = Math.max(numPartitions, new LLCSegmentName(segmentName).getPartitionId() + 1);
+ numPartitions = Math.max(numPartitions, new LLCSegmentName(segmentName).getPartitionGroupId() + 1);
}
}
return numPartitions;
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 2e73806..56ae29e 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
@@ -102,7 +102,7 @@ public class SegmentSizeBasedFlushThresholdUpdater implements FlushThresholdUpda
// less same characteristics at any one point in time).
// However, when we start a new table or change controller mastership, we can have any partition completing first.
// It is best to learn the ratio as quickly as we can, so we allow any partition to supply the value.
- if (new LLCSegmentName(newSegmentName).getPartitionId() == 0 || _latestSegmentRowsToSizeRatio == 0) {
+ if (new LLCSegmentName(newSegmentName).getPartitionGroupId() == 0 || _latestSegmentRowsToSizeRatio == 0) {
if (_latestSegmentRowsToSizeRatio > 0) {
_latestSegmentRowsToSizeRatio =
CURRENT_SEGMENT_RATIO_WEIGHT * currentRatio + PREVIOUS_SEGMENT_RATIO_WEIGHT * _latestSegmentRowsToSizeRatio;
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 4888f17..743e719 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
@@ -30,6 +30,8 @@ import java.util.Map;
import java.util.Random;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import javax.annotation.Nullable;
import org.apache.commons.io.FileUtils;
import org.apache.helix.model.IdealState;
@@ -48,6 +50,7 @@ import org.apache.pinot.controller.helix.core.assignment.segment.SegmentAssignme
import org.apache.pinot.controller.helix.core.realtime.segment.CommittingSegmentDescriptor;
import org.apache.pinot.controller.util.SegmentCompletionUtils;
import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
+import org.apache.pinot.core.realtime.impl.fakestream.FakePartitionGroupMetadata;
import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamConfigUtils;
import org.apache.pinot.core.segment.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -57,6 +60,7 @@ import org.apache.pinot.spi.env.PinotConfiguration;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+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.IngestionConfigUtils;
@@ -333,7 +337,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
assertTrue(oldSegmentZKMetadataMap.containsKey(segmentName));
assertTrue(segmentZKMetadataMap.containsKey(segmentName));
assertEquals(segmentZKMetadataMap.get(segmentName), oldSegmentZKMetadataMap.get(segmentName));
- oldNumPartitions = Math.max(oldNumPartitions, new LLCSegmentName(segmentName).getPartitionId() + 1);
+ oldNumPartitions = Math.max(oldNumPartitions, new LLCSegmentName(segmentName).getPartitionGroupId() + 1);
}
// Check that for new partitions, each partition should have exactly 1 new segment in CONSUMING state, and metadata
@@ -341,7 +345,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
Map<Integer, List<String>> partitionIdToSegmentsMap = new HashMap<>();
for (Map.Entry<String, Map<String, String>> entry : instanceStatesMap.entrySet()) {
String segmentName = entry.getKey();
- int partitionId = new LLCSegmentName(segmentName).getPartitionId();
+ int partitionId = new LLCSegmentName(segmentName).getPartitionGroupId();
partitionIdToSegmentsMap.computeIfAbsent(partitionId, k -> new ArrayList<>()).add(segmentName);
}
for (int partitionId = oldNumPartitions; partitionId < segmentManager._numPartitions; partitionId++) {
@@ -579,7 +583,7 @@ public class PinotLLCRealtimeSegmentManagerTest {
if (instanceStateMap.containsValue(SegmentStateModel.ONLINE) || instanceStateMap.containsValue(
SegmentStateModel.CONSUMING)) {
LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
- int partitionsId = llcSegmentName.getPartitionId();
+ int partitionsId = llcSegmentName.getPartitionGroupId();
Map<Integer, String> sequenceNumberToSegmentMap = partitionIdToSegmentsMap.get(partitionsId);
int sequenceNumber = llcSegmentName.getSequenceNumber();
assertFalse(sequenceNumberToSegmentMap.containsKey(sequenceNumber));
@@ -910,12 +914,12 @@ public class PinotLLCRealtimeSegmentManagerTest {
}
@Override
- int getNumPartitions(StreamConfig streamConfig) {
- return _numPartitions;
+ List<PartitionGroupMetadata> getPartitionGroupMetadataList(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
+ return IntStream.range(0, _numPartitions).mapToObj(FakePartitionGroupMetadata::new).collect(Collectors.toList());
}
@Override
- LongMsgOffset getPartitionOffset(StreamConfig streamConfig, OffsetCriteria offsetCriteria, int partitionId) {
+ LongMsgOffset getPartitionOffset(StreamConfig streamConfig, OffsetCriteria offsetCriteria, int partitionGroupId) {
// The criteria for this test should always be SMALLEST (for default streaming config and new added partitions)
assertTrue(offsetCriteria.isSmallest());
return PARTITION_OFFSET;
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 62e6d5b..f00facc 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
@@ -27,6 +27,7 @@ import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -214,7 +215,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// Semaphore for each partitionId only, which is to prevent two different Kafka consumers
// from consuming with the same partitionId in parallel in the same host.
// See the comments in {@link RealtimeTableDataManager}.
- private final Semaphore _partitionConsumerSemaphore;
+ private final Semaphore _partitionGroupConsumerSemaphore;
// A boolean flag to check whether the current thread has acquired the semaphore.
// This boolean is needed because the semaphore is shared by threads; every thread holding this semaphore can
// modify the permit. This boolean make sure the semaphore gets released only once when the partition stops consuming.
@@ -247,7 +248,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
private Thread _consumerThread;
private final String _streamTopic;
- private final int _streamPartitionId;
+ private final int _partitionGroupId;
final String _clientId;
private final LLCSegmentName _llcSegmentName;
private final RecordTransformer _recordTransformer;
@@ -704,7 +705,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
@Override
public Map<String, String> getPartitionToCurrentOffset() {
Map<String, String> partitionToCurrentOffset = new HashMap<>();
- partitionToCurrentOffset.put(String.valueOf(_streamPartitionId), _currentOffset.toString());
+ partitionToCurrentOffset.put(String.valueOf(_partitionGroupId), _currentOffset.toString());
return partitionToCurrentOffset;
}
@@ -729,8 +730,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
}
@VisibleForTesting
- protected Semaphore getPartitionConsumerSemaphore() {
- return _partitionConsumerSemaphore;
+ protected Semaphore getPartitionGroupConsumerSemaphore() {
+ return _partitionGroupConsumerSemaphore;
}
@VisibleForTesting
@@ -891,7 +892,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
closePartitionLevelConsumer();
closeStreamMetadataProvider();
if (_acquiredConsumerSemaphore.compareAndSet(true, false)) {
- _partitionConsumerSemaphore.release();
+ _partitionGroupConsumerSemaphore.release();
}
}
@@ -1101,7 +1102,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// If the transition is OFFLINE to ONLINE, the caller should have downloaded the segment and we don't reach here.
public LLRealtimeSegmentDataManager(RealtimeSegmentZKMetadata segmentZKMetadata, TableConfig tableConfig,
RealtimeTableDataManager realtimeTableDataManager, String resourceDataDir, IndexLoadingConfig indexLoadingConfig,
- Schema schema, LLCSegmentName llcSegmentName, Semaphore partitionConsumerSemaphore, ServerMetrics serverMetrics,
+ Schema schema, LLCSegmentName llcSegmentName, Semaphore partitionGroupConsumerSemaphore, ServerMetrics serverMetrics,
@Nullable PartitionUpsertMetadataManager partitionUpsertMetadataManager) {
_segBuildSemaphore = realtimeTableDataManager.getSegmentBuildSemaphore();
_segmentZKMetadata = (LLCRealtimeSegmentZKMetadata) segmentZKMetadata;
@@ -1128,10 +1129,10 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
_streamTopic = _partitionLevelStreamConfig.getTopicName();
_segmentNameStr = _segmentZKMetadata.getSegmentName();
_llcSegmentName = llcSegmentName;
- _streamPartitionId = _llcSegmentName.getPartitionId();
- _partitionConsumerSemaphore = partitionConsumerSemaphore;
+ _partitionGroupId = _llcSegmentName.getPartitionGroupId();
+ _partitionGroupConsumerSemaphore = partitionGroupConsumerSemaphore;
_acquiredConsumerSemaphore = new AtomicBoolean(false);
- _metricKeyName = _tableNameWithType + "-" + _streamTopic + "-" + _streamPartitionId;
+ _metricKeyName = _tableNameWithType + "-" + _streamTopic + "-" + _partitionGroupId;
segmentLogger = LoggerFactory.getLogger(LLRealtimeSegmentDataManager.class.getName() + "_" + _segmentNameStr);
_tableStreamName = _tableNameWithType + "_" + _streamTopic;
_memoryManager = getMemoryManager(realtimeTableDataManager.getConsumerDir(), _segmentNameStr,
@@ -1204,14 +1205,14 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// Create message decoder
Set<String> fieldsToRead = IngestionUtils.getFieldsForRecordExtractor(_tableConfig.getIngestionConfig(), _schema);
_messageDecoder = StreamDecoderProvider.create(_partitionLevelStreamConfig, fieldsToRead);
- _clientId = _streamTopic + "-" + _streamPartitionId;
+ _clientId = _streamTopic + "-" + _partitionGroupId;
// Create record transformer
_recordTransformer = CompositeTransformer.getDefaultTransformer(tableConfig, schema);
// Acquire semaphore to create Kafka consumers
try {
- _partitionConsumerSemaphore.acquire();
+ _partitionGroupConsumerSemaphore.acquire();
_acquiredConsumerSemaphore.set(true);
} catch (InterruptedException e) {
String errorMsg = "InterruptedException when acquiring the partitionConsumerSemaphore";
@@ -1237,7 +1238,8 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
// long as the partition function is not changed.
int numPartitions = columnPartitionConfig.getNumPartitions();
try {
- int numStreamPartitions = _streamMetadataProvider.fetchPartitionCount(/*maxWaitTimeMs=*/5000L);
+ // fixme: get this from ideal state
+ int numStreamPartitions = _streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000).size();
if (numStreamPartitions != numPartitions) {
segmentLogger.warn(
"Number of stream partitions: {} does not match number of partitions in the partition config: {}, using number of stream partitions",
@@ -1255,7 +1257,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
realtimeSegmentConfigBuilder.setPartitionColumn(partitionColumn);
realtimeSegmentConfigBuilder
.setPartitionFunction(PartitionFunctionFactory.getPartitionFunction(partitionFunctionName, numPartitions));
- realtimeSegmentConfigBuilder.setPartitionId(_streamPartitionId);
+ realtimeSegmentConfigBuilder.setPartitionId(_partitionGroupId);
} else {
segmentLogger.warn("Cannot partition on multiple columns: {}", columnPartitionMap.keySet());
}
@@ -1307,7 +1309,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
closePartitionLevelConsumer();
}
segmentLogger.info("Creating new stream consumer, reason: {}", reason);
- _partitionLevelConsumer = _streamConsumerFactory.createPartitionLevelConsumer(_clientId, _streamPartitionId);
+ _partitionLevelConsumer = _streamConsumerFactory.createPartitionLevelConsumer(_clientId, _partitionGroupId);
}
/**
@@ -1319,7 +1321,7 @@ public class LLRealtimeSegmentDataManager extends RealtimeSegmentDataManager {
closeStreamMetadataProvider();
}
segmentLogger.info("Creating new stream metadata provider, reason: {}", reason);
- _streamMetadataProvider = _streamConsumerFactory.createPartitionMetadataProvider(_clientId, _streamPartitionId);
+ _streamMetadataProvider = _streamConsumerFactory.createPartitionMetadataProvider(_clientId, _partitionGroupId);
}
// This should be done during commit? We may not always commit when we build a segment....
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 33283b9..9850048 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
@@ -36,6 +36,7 @@ import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
+import org.apache.helix.model.IdealState;
import org.apache.pinot.common.Utils;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.instance.InstanceZKMetadata;
@@ -89,7 +90,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
// In some streams, it's possible that having multiple consumers (with the same consumer name on the same host) consuming from the same stream partition can lead to bugs.
// The semaphores will stay in the hash map even if the consuming partitions move to a different host.
// We expect that there will be a small number of semaphores, but that may be ok.
- private final Map<Integer, Semaphore> _partitionIdToSemaphoreMap = new ConcurrentHashMap<>();
+ private final Map<Integer, Semaphore> _partitionGroupIdToSemaphoreMap = new ConcurrentHashMap<>();
// The old name of the stats file used to be stats.ser which we changed when we moved all packages
// from com.linkedin to org.apache because of not being able to deserialize the old files using the newer classes
@@ -274,7 +275,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
llcSegmentName = new LLCSegmentName(segmentName);
if (_tableUpsertMetadataManager != null) {
partitionUpsertMetadataManager =
- _tableUpsertMetadataManager.getOrCreatePartitionManager(llcSegmentName.getPartitionId());
+ _tableUpsertMetadataManager.getOrCreatePartitionManager(llcSegmentName.getPartitionGroupId());
}
}
@@ -307,11 +308,11 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
}
// Generates only one semaphore for every partitionId
- int partitionId = llcSegmentName.getPartitionId();
- _partitionIdToSemaphoreMap.putIfAbsent(partitionId, new Semaphore(1));
+ int partitionGroupId = llcSegmentName.getPartitionGroupId();
+ _partitionGroupIdToSemaphoreMap.putIfAbsent(partitionGroupId, new Semaphore(1));
manager =
new LLRealtimeSegmentDataManager(realtimeSegmentZKMetadata, tableConfig, this, _indexDir.getAbsolutePath(),
- indexLoadingConfig, schema, llcSegmentName, _partitionIdToSemaphoreMap.get(partitionId), _serverMetrics,
+ indexLoadingConfig, schema, llcSegmentName, _partitionGroupIdToSemaphoreMap.get(partitionGroupId), _serverMetrics,
partitionUpsertMetadataManager);
}
_logger.info("Initialize RealtimeSegmentDataManager - " + segmentName);
@@ -336,7 +337,7 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
columnToReaderMap.put(_timeColumnName, new PinotSegmentColumnReader(immutableSegment, _timeColumnName));
int numTotalDocs = immutableSegment.getSegmentMetadata().getTotalDocs();
String segmentName = immutableSegment.getSegmentName();
- int partitionId = new LLCSegmentName(immutableSegment.getSegmentName()).getPartitionId();
+ int partitionId = new LLCSegmentName(immutableSegment.getSegmentName()).getPartitionGroupId();
PartitionUpsertMetadataManager partitionUpsertMetadataManager =
_tableUpsertMetadataManager.getOrCreatePartitionManager(partitionId);
int numPrimaryKeyColumns = _primaryKeyColumns.size();
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
index 0017c43..d09bdeb 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManagerTest.java
@@ -719,7 +719,7 @@ public class LLRealtimeSegmentDataManagerTest {
long timeout = 10_000L;
FakeLLRealtimeSegmentDataManager firstSegmentDataManager = createFakeSegmentManager();
Assert.assertTrue(firstSegmentDataManager.getAcquiredConsumerSemaphore().get());
- Semaphore firstSemaphore = firstSegmentDataManager.getPartitionConsumerSemaphore();
+ Semaphore firstSemaphore = firstSegmentDataManager.getPartitionGroupConsumerSemaphore();
Assert.assertEquals(firstSemaphore.availablePermits(), 0);
Assert.assertFalse(firstSemaphore.hasQueuedThreads());
@@ -751,18 +751,18 @@ public class LLRealtimeSegmentDataManagerTest {
"Failed to acquire the semaphore for the second segment manager in " + timeout + "ms");
Assert.assertTrue(secondSegmentDataManager.get().getAcquiredConsumerSemaphore().get());
- Semaphore secondSemaphore = secondSegmentDataManager.get().getPartitionConsumerSemaphore();
+ Semaphore secondSemaphore = secondSegmentDataManager.get().getPartitionGroupConsumerSemaphore();
Assert.assertEquals(firstSemaphore, secondSemaphore);
Assert.assertEquals(secondSemaphore.availablePermits(), 0);
Assert.assertFalse(secondSemaphore.hasQueuedThreads());
// Call destroy method the 2nd time on the first segment manager, the permits in semaphore won't increase.
firstSegmentDataManager.destroy();
- Assert.assertEquals(firstSegmentDataManager.getPartitionConsumerSemaphore().availablePermits(), 0);
+ Assert.assertEquals(firstSegmentDataManager.getPartitionGroupConsumerSemaphore().availablePermits(), 0);
// The permit finally gets released in the Semaphore.
secondSegmentDataManager.get().destroy();
- Assert.assertEquals(secondSegmentDataManager.get().getPartitionConsumerSemaphore().availablePermits(), 1);
+ Assert.assertEquals(secondSegmentDataManager.get().getPartitionGroupConsumerSemaphore().availablePermits(), 1);
}
public static class FakeLLRealtimeSegmentDataManager extends LLRealtimeSegmentDataManager {
@@ -800,7 +800,7 @@ public class LLRealtimeSegmentDataManagerTest {
throws Exception {
super(segmentZKMetadata, tableConfig, realtimeTableDataManager, resourceDataDir,
new IndexLoadingConfig(makeInstanceDataManagerConfig(), tableConfig), schema, llcSegmentName,
- semaphoreMap.get(llcSegmentName.getPartitionId()), serverMetrics,
+ semaphoreMap.get(llcSegmentName.getPartitionGroupId()), serverMetrics,
new PartitionUpsertMetadataManager("testTable_REALTIME", 0, serverMetrics));
_state = LLRealtimeSegmentDataManager.class.getDeclaredField("_state");
_state.setAccessible(true);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java
new file mode 100644
index 0000000..78ee12c
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakePartitionGroupMetadata.java
@@ -0,0 +1,48 @@
+package org.apache.pinot.core.realtime.impl.fakestream;
+
+import org.apache.pinot.spi.stream.Checkpoint;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+
+
+public class FakePartitionGroupMetadata implements PartitionGroupMetadata {
+
+ private final int _groupId;
+ public FakePartitionGroupMetadata(int groupId) {
+ _groupId = groupId;
+ }
+
+ @Override
+ public int getGroupId() {
+ return getGroupId();
+ }
+
+ @Override
+ public Checkpoint getStartCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public Checkpoint getEndCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public void setStartCheckpoint(Checkpoint startCheckpoint) {
+
+ }
+
+ @Override
+ public void setEndCheckpoint(Checkpoint endCheckpoint) {
+
+ }
+
+ @Override
+ public byte[] serialize() {
+ return new byte[0];
+ }
+
+ @Override
+ public PartitionGroupMetadata deserialize(byte[] blob) {
+ return null;
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
index 9669223..289b226 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.realtime.impl.fakestream;
-import java.util.List;
import java.util.Set;
import org.apache.pinot.core.util.IngestionUtils;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -69,14 +68,9 @@ public class FakeStreamConsumerFactory extends StreamConsumerFactory {
return new FakeStreamMetadataProvider(_streamConfig);
}
- @Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
- return null;
- }
@Override
- public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ public PartitionGroupConsumer createPartitionGroupConsumer(PartitionGroupMetadata metadata) {
return null;
}
@@ -93,7 +87,7 @@ public class FakeStreamConsumerFactory extends StreamConsumerFactory {
// stream metadata provider
StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(clientId);
- int partitionCount = streamMetadataProvider.fetchPartitionCount(10_000);
+ int partitionCount = streamMetadataProvider.getPartitionGroupMetadataList(null, 10_000).size();
System.out.println(partitionCount);
// Partition metadata provider
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
index e0b8ebd..c96d06a 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamMetadataProvider.java
@@ -19,9 +19,12 @@
package org.apache.pinot.core.realtime.impl.fakestream;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nonnull;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
@@ -31,7 +34,7 @@ import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
* StreamMetadataProvider implementation for the fake stream
*/
public class FakeStreamMetadataProvider implements StreamMetadataProvider {
- private int _numPartitions;
+ private final int _numPartitions;
public FakeStreamMetadataProvider(StreamConfig streamConfig) {
_numPartitions = FakeStreamConfigUtils.getNumPartitions(streamConfig);
@@ -42,6 +45,16 @@ public class FakeStreamMetadataProvider implements StreamMetadataProvider {
return _numPartitions;
}
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
+ List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>();
+ for (int i = 0; i < _numPartitions; i++) {
+ partitionGroupMetadataList.add(new FakePartitionGroupMetadata(i));
+ }
+ return partitionGroupMetadataList;
+ }
+
public long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis) throws TimeoutException {
throw new UnsupportedOperationException("This method is deprecated");
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
index 808a464..d917d73 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/FlakyConsumerRealtimeClusterIntegrationTest.java
@@ -19,7 +19,6 @@
package org.apache.pinot.integration.tests;
import java.lang.reflect.Constructor;
-import java.util.List;
import java.util.Random;
import java.util.Set;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -122,13 +121,7 @@ public class FlakyConsumerRealtimeClusterIntegrationTest extends RealtimeCluster
}
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
- return null;
- }
-
- @Override
- public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ public PartitionGroupConsumer createPartitionGroupConsumer(PartitionGroupMetadata metadata) {
return null;
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentPartitionLLCRealtimeClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentPartitionLLCRealtimeClusterIntegrationTest.java
index cd4f9b3..0196bde 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentPartitionLLCRealtimeClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/SegmentPartitionLLCRealtimeClusterIntegrationTest.java
@@ -165,7 +165,7 @@ public class SegmentPartitionLLCRealtimeClusterIntegrationTest extends BaseClust
assertNotNull(columnPartitionMetadata);
assertTrue(columnPartitionMetadata.getFunctionName().equalsIgnoreCase("murmur"));
assertEquals(columnPartitionMetadata.getNumPartitions(), 2);
- int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionId();
+ int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionGroupId();
assertEquals(columnPartitionMetadata.getPartitions(), Collections.singleton(streamPartitionId));
numSegmentsForPartition[streamPartitionId]++;
}
@@ -236,7 +236,7 @@ public class SegmentPartitionLLCRealtimeClusterIntegrationTest extends BaseClust
assertNotNull(columnPartitionMetadata);
assertTrue(columnPartitionMetadata.getFunctionName().equalsIgnoreCase("murmur"));
assertEquals(columnPartitionMetadata.getNumPartitions(), 2);
- int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionId();
+ int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionGroupId();
numSegmentsForPartition[streamPartitionId]++;
if (segmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
@@ -313,7 +313,7 @@ public class SegmentPartitionLLCRealtimeClusterIntegrationTest extends BaseClust
assertNotNull(columnPartitionMetadata);
assertTrue(columnPartitionMetadata.getFunctionName().equalsIgnoreCase("murmur"));
assertEquals(columnPartitionMetadata.getNumPartitions(), 2);
- int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionId();
+ int streamPartitionId = new LLCSegmentName(segmentZKMetadata.getSegmentName()).getPartitionGroupId();
numSegmentsForPartition[streamPartitionId]++;
if (segmentZKMetadata.getStatus() == Status.IN_PROGRESS) {
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
index b8ed19d..82c282c 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaConsumerFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.plugin.stream.kafka09;
-import java.util.List;
import java.util.Set;
import org.apache.pinot.spi.stream.PartitionGroupConsumer;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
@@ -55,13 +54,7 @@ public class KafkaConsumerFactory extends StreamConsumerFactory {
}
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
- return null;
- }
-
- @Override
- public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ public PartitionGroupConsumer createPartitionGroupConsumer(PartitionGroupMetadata metadata) {
return null;
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java
new file mode 100644
index 0000000..1d792ac
--- /dev/null
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionGroupMetadata.java
@@ -0,0 +1,48 @@
+package org.apache.pinot.plugin.stream.kafka09;
+
+import org.apache.pinot.spi.stream.Checkpoint;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+
+
+public class KafkaPartitionGroupMetadata implements PartitionGroupMetadata {
+
+ private final int _groupId;
+ public KafkaPartitionGroupMetadata(int partitionId) {
+ _groupId = partitionId;
+ }
+
+ @Override
+ public int getGroupId() {
+ return _groupId;
+ }
+
+ @Override
+ public Checkpoint getStartCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public Checkpoint getEndCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public void setStartCheckpoint(Checkpoint startCheckpoint) {
+
+ }
+
+ @Override
+ public void setEndCheckpoint(Checkpoint endCheckpoint) {
+
+ }
+
+ @Override
+ public byte[] serialize() {
+ return new byte[0];
+ }
+
+ @Override
+ public PartitionGroupMetadata deserialize(byte[] blob) {
+ return null;
+ }
+}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
index 06ee697..865ae96 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/main/java/org/apache/pinot/plugin/stream/kafka09/KafkaStreamMetadataProvider.java
@@ -22,9 +22,13 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.Uninterruptibles;
import java.io.IOException;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.TopicAndPartition;
import kafka.javaapi.OffsetRequest;
@@ -36,6 +40,7 @@ import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.protocol.Errors;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
@@ -84,7 +89,12 @@ public class KafkaStreamMetadataProvider extends KafkaConnectionHandler implemen
* @return
*/
@Override
+ @Deprecated
public synchronized int fetchPartitionCount(long timeoutMillis) {
+ return fetchPartitionCountInternal(timeoutMillis);
+ }
+
+ private int fetchPartitionCountInternal(long timeoutMillis) {
int unknownTopicReplyCount = 0;
final int MAX_UNKNOWN_TOPIC_REPLY_COUNT = 10;
int kafkaErrorCount = 0;
@@ -145,6 +155,22 @@ public class KafkaStreamMetadataProvider extends KafkaConnectionHandler implemen
throw new TimeoutException();
}
+ /**
+ * Fetch the partition group metadata list
+ * @param currentPartitionGroupsMetadata In case of Kafka, each partition group contains a single partition.
+ * Hence current partition groups are not needed to compute the new partition groups
+ */
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ @Nullable List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
+ int partitionCount = fetchPartitionCountInternal(timeoutMillis);
+ List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>(partitionCount);
+ for (int i = 0; i < partitionCount; i++) {
+ partitionGroupMetadataList.add(new KafkaPartitionGroupMetadata(i));
+ }
+ return partitionGroupMetadataList;
+ }
+
public synchronized long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis)
throws java.util.concurrent.TimeoutException {
throw new UnsupportedOperationException("The use of this method s not supported");
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
index beb82e5..fbdfdfb 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
@@ -291,7 +291,7 @@ public class KafkaPartitionLevelConsumerTest {
KafkaStreamMetadataProvider streamMetadataProvider =
new KafkaStreamMetadataProvider(clientId, streamConfig, mockKafkaSimpleConsumerFactory);
- Assert.assertEquals(streamMetadataProvider.fetchPartitionCount(10000L), 2);
+ Assert.assertEquals(streamMetadataProvider.getPartitionGroupMetadataList(null, 10000L), 2);
}
@Test
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
index 806baff..c73aacb 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaConsumerFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.plugin.stream.kafka20;
-import java.util.List;
import java.util.Set;
import org.apache.pinot.spi.stream.PartitionGroupConsumer;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
@@ -52,13 +51,7 @@ public class KafkaConsumerFactory extends StreamConsumerFactory {
}
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata) {
- return null;
- }
-
- @Override
- public PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata) {
+ public PartitionGroupConsumer createPartitionGroupConsumer(PartitionGroupMetadata metadata) {
return null;
}
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java
new file mode 100644
index 0000000..31ae75a
--- /dev/null
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionGroupMetadata.java
@@ -0,0 +1,48 @@
+package org.apache.pinot.plugin.stream.kafka20;
+
+import org.apache.pinot.spi.stream.Checkpoint;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+
+
+public class KafkaPartitionGroupMetadata implements PartitionGroupMetadata {
+
+ private final int _groupId;
+ public KafkaPartitionGroupMetadata(int partitionId) {
+ _groupId = partitionId;
+ }
+
+ @Override
+ public int getGroupId() {
+ return _groupId;
+ }
+
+ @Override
+ public Checkpoint getStartCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public Checkpoint getEndCheckpoint() {
+ return null;
+ }
+
+ @Override
+ public void setStartCheckpoint(Checkpoint startCheckpoint) {
+
+ }
+
+ @Override
+ public void setEndCheckpoint(Checkpoint endCheckpoint) {
+
+ }
+
+ @Override
+ public byte[] serialize() {
+ return new byte[0];
+ }
+
+ @Override
+ public PartitionGroupMetadata deserialize(byte[] blob) {
+ return null;
+ }
+}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
index c0e2041..187c61b 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
@@ -21,11 +21,15 @@ package org.apache.pinot.plugin.stream.kafka20;
import com.google.common.base.Preconditions;
import java.io.IOException;
import java.time.Duration;
+import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
@@ -42,10 +46,27 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
}
@Override
+ @Deprecated
public int fetchPartitionCount(long timeoutMillis) {
return _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis)).size();
}
+ /**
+ * Fetch the partitionGroupMetadata list.
+ * @param currentPartitionGroupsMetadata In case of Kafka, each partition group contains a single partition.
+ * Hence current partition groups are not needed to compute the new partition groups
+ */
+ @Override
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ @Nullable List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
+ int partitionCount = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis)).size();
+ List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>(partitionCount);
+ for (int i = 0; i < partitionCount; i++) {
+ partitionGroupMetadataList.add(new KafkaPartitionGroupMetadata(i));
+ }
+ return partitionGroupMetadataList;
+ }
+
public synchronized long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis)
throws java.util.concurrent.TimeoutException {
throw new UnsupportedOperationException("The use of this method is not supported");
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionCountFetcher.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
similarity index 74%
rename from pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionCountFetcher.java
rename to pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
index d523235..e1ce1a6 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionCountFetcher.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupMetadataFetcher.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.spi.stream;
+import java.util.List;
import java.util.concurrent.Callable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -26,24 +27,27 @@ import org.slf4j.LoggerFactory;
/**
* Fetches the partition count of a stream using the {@link StreamMetadataProvider}
*/
-public class PartitionCountFetcher implements Callable<Boolean> {
+public class PartitionGroupMetadataFetcher implements Callable<Boolean> {
- private static final Logger LOGGER = LoggerFactory.getLogger(PartitionCountFetcher.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(PartitionGroupMetadataFetcher.class);
private int _partitionCount = -1;
+ private List<PartitionGroupMetadata> _partitionGroupMetadataList;
+ private List<PartitionGroupMetadata> _currentPartitionGroupMetadata;
private final StreamConfig _streamConfig;
private StreamConsumerFactory _streamConsumerFactory;
private Exception _exception;
private final String _topicName;
- public PartitionCountFetcher(StreamConfig streamConfig) {
+ public PartitionGroupMetadataFetcher(StreamConfig streamConfig, List<PartitionGroupMetadata> currentPartitionGroupMetadataList) {
_streamConfig = streamConfig;
_streamConsumerFactory = StreamConsumerFactoryProvider.create(_streamConfig);
_topicName = streamConfig.getTopicName();
+ _currentPartitionGroupMetadata = currentPartitionGroupMetadataList;
}
- public int getPartitionCount() {
- return _partitionCount;
+ public List<PartitionGroupMetadata> getPartitionGroupMetadataList() {
+ return _partitionGroupMetadataList;
}
public Exception getException() {
@@ -59,10 +63,10 @@ public class PartitionCountFetcher implements Callable<Boolean> {
public Boolean call()
throws Exception {
- String clientId = PartitionCountFetcher.class.getSimpleName() + "-" + _topicName;
+ String clientId = PartitionGroupMetadataFetcher.class.getSimpleName() + "-" + _topicName;
try (
StreamMetadataProvider streamMetadataProvider = _streamConsumerFactory.createStreamMetadataProvider(clientId)) {
- _partitionCount = streamMetadataProvider.fetchPartitionCount(/*maxWaitTimeMs=*/5000L);
+ _partitionGroupMetadataList = streamMetadataProvider.getPartitionGroupMetadataList(_currentPartitionGroupMetadata, /*maxWaitTimeMs=*/5000L);
if (_exception != null) {
// We had at least one failure, but succeeded now. Log an info
LOGGER.info("Successfully retrieved partition count as {} for topic {}", _partitionCount, _topicName);
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionOffsetFetcher.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionOffsetFetcher.java
index 1d50160..b92f04d 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionOffsetFetcher.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionOffsetFetcher.java
@@ -33,16 +33,16 @@ public class PartitionOffsetFetcher implements Callable<Boolean> {
private final String _topicName;
private final OffsetCriteria _offsetCriteria;
- private final int _partitionId;
+ private final int _partitionGroupId;
private Exception _exception = null;
private StreamPartitionMsgOffset _offset;
private StreamConsumerFactory _streamConsumerFactory;
StreamConfig _streamConfig;
- public PartitionOffsetFetcher(final OffsetCriteria offsetCriteria, int partitionId, StreamConfig streamConfig) {
+ public PartitionOffsetFetcher(final OffsetCriteria offsetCriteria, int partitionGroupId, StreamConfig streamConfig) {
_offsetCriteria = offsetCriteria;
- _partitionId = partitionId;
+ _partitionGroupId = partitionGroupId;
_streamConfig = streamConfig;
_streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
_topicName = streamConfig.getTopicName();
@@ -64,18 +64,19 @@ public class PartitionOffsetFetcher implements Callable<Boolean> {
@Override
public Boolean call()
throws Exception {
- String clientId = PartitionOffsetFetcher.class.getSimpleName() + "-" + _topicName + "-" + _partitionId;
+ String clientId = PartitionOffsetFetcher.class.getSimpleName() + "-" + _topicName + "-" + _partitionGroupId;
try (StreamMetadataProvider streamMetadataProvider = _streamConsumerFactory
- .createPartitionMetadataProvider(clientId, _partitionId)) {
+ .createPartitionMetadataProvider(clientId, _partitionGroupId)) {
_offset =
streamMetadataProvider.fetchStreamPartitionOffset(_offsetCriteria, STREAM_PARTITION_OFFSET_FETCH_TIMEOUT_MILLIS);
if (_exception != null) {
LOGGER.info("Successfully retrieved offset({}) for stream topic {} partition {}", _offset, _topicName,
- _partitionId);
+ _partitionGroupId);
}
return Boolean.TRUE;
} catch (TransientConsumerException e) {
- LOGGER.warn("Temporary exception when fetching offset for topic {} partition {}:{}", _topicName, _partitionId,
+ LOGGER.warn("Temporary exception when fetching offset for topic {} partition {}:{}", _topicName,
+ _partitionGroupId,
e.getMessage());
_exception = e;
return Boolean.FALSE;
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
index 4db0fb1..9caf61b 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamConsumerFactory.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.spi.stream;
-import java.util.List;
import java.util.Set;
@@ -42,6 +41,7 @@ public abstract class StreamConsumerFactory {
* @param partition the partition id of the partition for which this consumer is being created
* @return
*/
+ @Deprecated
public abstract PartitionLevelConsumer createPartitionLevelConsumer(String clientId, int partition);
/**
@@ -74,10 +74,6 @@ public abstract class StreamConsumerFactory {
return new LongMsgOffsetFactory();
}
- // takes the current state of partition groups (groupings of shards, the state of the consumption) and creates the new state
- public abstract List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata);
-
// creates a consumer which consumes from a partition group
- public abstract PartitionGroupConsumer createConsumer(PartitionGroupMetadata metadata);
+ public abstract PartitionGroupConsumer createPartitionGroupConsumer(PartitionGroupMetadata metadata);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
index 557ffc4..5b9104e 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
@@ -19,6 +19,7 @@
package org.apache.pinot.spi.stream;
import java.io.Closeable;
+import java.util.List;
import javax.annotation.Nonnull;
import org.apache.pinot.spi.annotations.InterfaceAudience;
import org.apache.pinot.spi.annotations.InterfaceStability;
@@ -32,11 +33,15 @@ import org.apache.pinot.spi.annotations.InterfaceStability;
public interface StreamMetadataProvider extends Closeable {
/**
* Fetches the number of partitions for a topic given the stream configs
- * @param timeoutMillis
- * @return
+ * @deprecated use getPartitionGroupMetadataList instead
*/
+ @Deprecated
int fetchPartitionCount(long timeoutMillis);
+ // takes the current state of partition groups (groupings of shards, the state of the consumption) and creates the new state
+ List<PartitionGroupMetadata> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis);
+
// Issue 5953 Retain this interface for 0.5.0, remove in 0.6.0
@Deprecated
long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 05/08: Separate PartitionGroupInfo and
PartitionGroupMetadata
Posted by ne...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nehapawar pushed a commit to branch sharded_consumer_type_support
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 1f81c1eaf2eab7eb10a6112082a4efd028385501
Author: Neha Pawar <ne...@gmail.com>
AuthorDate: Thu Dec 31 14:46:46 2020 -0800
Separate PartitionGroupInfo and PartitionGroupMetadata
---
.../helix/core/PinotHelixResourceManager.java | 12 +--
.../realtime/PinotLLCRealtimeSegmentManager.java | 108 ++++++++-------------
.../impl/fakestream/FakeStreamConsumerFactory.java | 3 +-
.../kafka09/KafkaPartitionLevelConsumerTest.java | 2 +-
.../kafka20/KafkaStreamMetadataProvider.java | 28 ++++--
.../pinot/spi/stream/PartitionGroupInfo.java | 43 ++++++++
.../pinot/spi/stream/StreamMetadataProvider.java | 6 +-
7 files changed, 115 insertions(+), 87 deletions(-)
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 1ff3f9d..ebfbfa1 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
@@ -122,12 +122,7 @@ import org.apache.pinot.spi.config.table.TenantConfig;
import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.config.tenant.Tenant;
import org.apache.pinot.spi.data.Schema;
-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.StreamConsumerFactory;
-import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
-import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.spi.utils.retry.RetryPolicies;
@@ -1342,9 +1337,8 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
- LOGGER.info("Successfully added Helix entries for low-level consumers for {} ", realtimeTableName);
- _pinotLLCRealtimeSegmentManager.setupNewShardedTable(rawRealtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setupNewShardedTable(realtimeTableConfig, idealState);
+ LOGGER.info("Successfully setup table for SHARDED consumers for {} ", realtimeTableName);
} else {
if (streamConfig.hasHighLevelConsumerType()) {
@@ -1372,7 +1366,7 @@ public class PinotHelixResourceManager {
idealState = PinotTableIdealStateBuilder
.buildLowLevelRealtimeIdealStateFor(realtimeTableName, realtimeTableConfig, idealState,
_enableBatchMessageMode);
- _pinotLLCRealtimeSegmentManager.setUpNewTable(realtimeTableConfig, idealState);
+ _pinotLLCRealtimeSegmentManager.setupNewShardedTable(realtimeTableConfig, idealState);
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);
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 0de5e91..423a0b2 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
@@ -29,6 +29,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -78,6 +79,7 @@ import org.apache.pinot.spi.config.table.assignment.InstancePartitionsType;
import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.PartitionLevelStreamConfig;
import org.apache.pinot.spi.stream.PartitionOffsetFetcher;
@@ -221,8 +223,14 @@ public class PinotLLCRealtimeSegmentManager {
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
.createStreamMetadataProvider(streamConfig.getTopicName() + "_" + System.currentTimeMillis());
- List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000);
+
+ List<PartitionGroupInfo> newPartitionGroupMetadataList;
+ try {
+ newPartitionGroupMetadataList =
+ streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 5000);
+ } catch (TimeoutException e) {
+ throw new IllegalStateException(e);
+ }
int numPartitionGroups = newPartitionGroupMetadataList.size();
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
@@ -234,8 +242,8 @@ public class PinotLLCRealtimeSegmentManager {
long currentTimeMs = getCurrentTimeMs();
Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
- for (PartitionGroupMetadata partitionGroupMetadata : newPartitionGroupMetadataList) {
- String segmentName = setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupMetadata.getPartitionGroupId(),
+ for (PartitionGroupInfo partitionGroupInfo : newPartitionGroupMetadataList) {
+ String segmentName = setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupInfo,
currentTimeMs, instancePartitions, numPartitionGroups, numReplicas);
updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment,
instancePartitionsMap);
@@ -277,50 +285,6 @@ public class PinotLLCRealtimeSegmentManager {
}
/**
- * Sets up the initial segments for a new LLC real-time table.
- * <p>NOTE: the passed in IdealState may contain HLC segments if both HLC and LLC are configured.
- */
- public void setUpNewTable(TableConfig tableConfig, IdealState idealState) {
- Preconditions.checkState(!_isStopping, "Segment manager is stopping");
-
- 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));
- InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
- List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
- int numPartitionGroups = getPartitionGroupMetadataList(streamConfig, currentPartitionGroupMetadataList).size();
- int numReplicas = getNumReplicas(tableConfig, instancePartitions);
-
- SegmentAssignment segmentAssignment = SegmentAssignmentFactory.getSegmentAssignment(_helixManager, tableConfig);
- Map<InstancePartitionsType, InstancePartitions> instancePartitionsMap =
- Collections.singletonMap(InstancePartitionsType.CONSUMING, instancePartitions);
-
- long currentTimeMs = getCurrentTimeMs();
- Map<String, Map<String, String>> instanceStatesMap = idealState.getRecord().getMapFields();
- for (int partitionGroupId = 0; partitionGroupId < numPartitionGroups; partitionGroupId++) {
- String segmentName =
- setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupId, currentTimeMs, instancePartitions, numPartitionGroups,
- numReplicas);
- updateInstanceStatesForNewConsumingSegment(instanceStatesMap, null, segmentName, segmentAssignment,
- instancePartitionsMap);
- }
-
- setIdealState(realtimeTableName, idealState);
- }
-
- /**
* Removes all LLC segments from the given IdealState.
*/
public void removeLLCSegments(IdealState idealState) {
@@ -538,15 +502,23 @@ public class PinotLLCRealtimeSegmentManager {
// Step-2
// Say we currently were consuming from 3 shards A, B, C. Of those, A is the one committing. Also suppose that new partition D has come up
+
// get current partition groups - this gives current state of latest segments for each partition [A - DONE], [B - IN_PROGRESS], [C - IN_PROGRESS]
List<PartitionGroupMetadata> currentPartitionGroupMetadataList = getCurrentPartitionGroupMetadataList(idealState);
- StreamConfig streamConfig = new StreamConfig(tableConfig.getTableName(), IngestionConfigUtils.getStreamConfigMap(tableConfig));
+ PartitionLevelStreamConfig streamConfig = new PartitionLevelStreamConfig(tableConfig.getTableName(),
+ IngestionConfigUtils.getStreamConfigMap(tableConfig));
StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
StreamMetadataProvider streamMetadataProvider = streamConsumerFactory
.createStreamMetadataProvider(streamConfig.getTopicName() + " " + System.currentTimeMillis());
+
// find new partition groups [A],[B],[C],[D]
- List<PartitionGroupMetadata> newPartitionGroupMetadataList =
- streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
+ List<PartitionGroupInfo> newPartitionGroupMetadataList;
+ try {
+ newPartitionGroupMetadataList =
+ streamMetadataProvider.getPartitionGroupMetadataList(currentPartitionGroupMetadataList, 1000);
+ } catch (TimeoutException e) {
+ throw new IllegalStateException(e);
+ }
// create new segment metadata, only if it is not IN_PROGRESS in the current state
Map<Integer, PartitionGroupMetadata> currentGroupIdToMetadata = currentPartitionGroupMetadataList.stream().collect(
@@ -555,25 +527,24 @@ public class PinotLLCRealtimeSegmentManager {
List<String> newConsumingSegmentNames = new ArrayList<>();
String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
long newSegmentCreationTimeMs = getCurrentTimeMs();
- for (PartitionGroupMetadata partitionGroupMetadata : newPartitionGroupMetadataList) {
- int newPartitionGroupId = partitionGroupMetadata.getPartitionGroupId();
+ for (PartitionGroupInfo partitionGroupInfo : newPartitionGroupMetadataList) {
+ int newPartitionGroupId = partitionGroupInfo.getPartitionGroupId();
PartitionGroupMetadata currentPartitionGroupMetadata = currentGroupIdToMetadata.get(newPartitionGroupId);
- if (currentPartitionGroupMetadata == null) { // not present in current state
+ if (currentPartitionGroupMetadata == null) { // not present in current state. New partition found.
// make new segment
- LLCSegmentName newLLCSegmentName =
- new LLCSegmentName(rawTableName, newPartitionGroupId, STARTING_SEQUENCE_NUMBER, newSegmentCreationTimeMs);
- createNewSegmentZKMetadata(tableConfig, new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig)), newLLCSegmentName, newSegmentCreationTimeMs,
- committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, numPartitions, numReplicas);
- newConsumingSegmentNames.add(newLLCSegmentName.getSegmentName());
+ String newLLCSegmentName =
+ setupNewPartitionGroup(tableConfig, streamConfig, partitionGroupInfo, newSegmentCreationTimeMs,
+ instancePartitions, numPartitions, numReplicas);
+ newConsumingSegmentNames.add(newLLCSegmentName);
} else {
String currentStatus = currentPartitionGroupMetadata.getStatus();
- if (!currentStatus.equals(Status.IN_PROGRESS.toString())) { // not IN_PROGRESS anymore in current state
- // make new segment
+ if (!currentStatus.equals(Status.IN_PROGRESS.toString())) {
+ // not IN_PROGRESS anymore in current state. Should be DONE.
+ // This should ONLY happen for the committing segment's partition. Need to trigger new consuming segment
+ // todo: skip this if the partition doesn't match with the committing segment?
LLCSegmentName newLLCSegmentName = new LLCSegmentName(rawTableName, newPartitionGroupId,
currentPartitionGroupMetadata.getSequenceNumber() + 1, newSegmentCreationTimeMs);
- createNewSegmentZKMetadata(tableConfig, new PartitionLevelStreamConfig(tableConfig.getTableName(),
- IngestionConfigUtils.getStreamConfigMap(tableConfig)), newLLCSegmentName, newSegmentCreationTimeMs,
+ createNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegmentName, newSegmentCreationTimeMs,
committingSegmentDescriptor, committingSegmentZKMetadata, instancePartitions, numPartitions, numReplicas);
newConsumingSegmentNames.add(newLLCSegmentName.getSegmentName());
}
@@ -1162,19 +1133,20 @@ public class PinotLLCRealtimeSegmentManager {
* Sets up a new partition.
* <p>Persists the ZK metadata for the first CONSUMING segment, and returns the segment name.
*/
- private String setupNewPartitionGroup(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig, int partitionGroupId,
+ private String setupNewPartitionGroup(TableConfig tableConfig, PartitionLevelStreamConfig streamConfig, PartitionGroupInfo partitionGroupInfo,
long creationTimeMs, InstancePartitions instancePartitions, int numPartitionGroups, int numReplicas) {
String realtimeTableName = tableConfig.getTableName();
+ int partitionGroupId = partitionGroupInfo.getPartitionGroupId();
+ String startCheckpoint = partitionGroupInfo.getStartCheckpoint();
LOGGER.info("Setting up new partition group: {} for table: {}", partitionGroupId, realtimeTableName);
String rawTableName = TableNameBuilder.extractRawTableName(realtimeTableName);
LLCSegmentName newLLCSegmentName =
new LLCSegmentName(rawTableName, partitionGroupId, STARTING_SEQUENCE_NUMBER, creationTimeMs);
String newSegmentName = newLLCSegmentName.getSegmentName();
- StreamPartitionMsgOffset startOffset =
- getPartitionOffset(streamConfig, streamConfig.getOffsetCriteria(), partitionGroupId);
+
CommittingSegmentDescriptor committingSegmentDescriptor =
- new CommittingSegmentDescriptor(null, startOffset.toString(), 0);
+ new CommittingSegmentDescriptor(null, startCheckpoint, 0);
createNewSegmentZKMetadata(tableConfig, streamConfig, newLLCSegmentName, creationTimeMs,
committingSegmentDescriptor, null, instancePartitions, numPartitionGroups, numReplicas);
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
index 289b226..54be1b6 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/realtime/impl/fakestream/FakeStreamConsumerFactory.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.realtime.impl.fakestream;
+import java.util.Collections;
import java.util.Set;
import org.apache.pinot.core.util.IngestionUtils;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -87,7 +88,7 @@ public class FakeStreamConsumerFactory extends StreamConsumerFactory {
// stream metadata provider
StreamMetadataProvider streamMetadataProvider = streamConsumerFactory.createStreamMetadataProvider(clientId);
- int partitionCount = streamMetadataProvider.getPartitionGroupMetadataList(null, 10_000).size();
+ int partitionCount = streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 10_000).size();
System.out.println(partitionCount);
// Partition metadata provider
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
index fbdfdfb..43b72a8 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-0.9/src/test/java/org/apache/pinot/plugin/stream/kafka09/KafkaPartitionLevelConsumerTest.java
@@ -291,7 +291,7 @@ public class KafkaPartitionLevelConsumerTest {
KafkaStreamMetadataProvider streamMetadataProvider =
new KafkaStreamMetadataProvider(clientId, streamConfig, mockKafkaSimpleConsumerFactory);
- Assert.assertEquals(streamMetadataProvider.getPartitionGroupMetadataList(null, 10000L), 2);
+ Assert.assertEquals(streamMetadataProvider.getPartitionGroupMetadataList(Collections.emptyList(), 10000L), 2);
}
@Test
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
index 187c61b..eb606f2 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java
@@ -29,6 +29,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.OffsetCriteria;
+import org.apache.pinot.spi.stream.PartitionGroupInfo;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
@@ -37,12 +38,15 @@ import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHandler implements StreamMetadataProvider {
+ private StreamConfig _streamConfig;
+
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ _streamConfig = streamConfig;
}
@Override
@@ -57,14 +61,26 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
* Hence current partition groups are not needed to compute the new partition groups
*/
@Override
- public List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- @Nullable List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis) {
+ public List<PartitionGroupInfo> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
+ throws TimeoutException {
int partitionCount = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis)).size();
- List<PartitionGroupMetadata> partitionGroupMetadataList = new ArrayList<>(partitionCount);
- for (int i = 0; i < partitionCount; i++) {
- partitionGroupMetadataList.add(new KafkaPartitionGroupMetadata(i));
+ List<PartitionGroupInfo> newPartitionGroupInfoList = new ArrayList<>(partitionCount);
+
+ // add a PartitionGroupInfo into the list foreach partition already present in current.
+ // the end checkpoint is set as checkpoint
+ for (PartitionGroupMetadata currentPartitionGroupMetadata : currentPartitionGroupsMetadata) {
+ newPartitionGroupInfoList.add(new PartitionGroupInfo(currentPartitionGroupMetadata.getPartitionGroupId(),
+ currentPartitionGroupMetadata.getEndCheckpoint()));
+ }
+ // add PartitiongroupInfo for new partitions
+ // use offset criteria from stream config
+ for (int i = currentPartitionGroupsMetadata.size(); i < partitionCount; i++) {
+ StreamPartitionMsgOffset streamPartitionMsgOffset =
+ fetchStreamPartitionOffset(_streamConfig.getOffsetCriteria(), 5000);
+ newPartitionGroupInfoList.add(new PartitionGroupInfo(i, streamPartitionMsgOffset.toString()));
}
- return partitionGroupMetadataList;
+ return newPartitionGroupInfoList;
}
public synchronized long fetchPartitionOffset(@Nonnull OffsetCriteria offsetCriteria, long timeoutMillis)
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfo.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfo.java
new file mode 100644
index 0000000..438e148
--- /dev/null
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/PartitionGroupInfo.java
@@ -0,0 +1,43 @@
+/**
+ * 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.spi.stream;
+
+public class PartitionGroupInfo {
+
+ // fixme: Make partitionGroupId string everywhere (LLCSegmentName, StreamMetadataProvider)
+ private final int _partitionGroupId;
+ private String _startCheckpoint;
+
+ public PartitionGroupInfo(int partitionGroupId, String startCheckpoint) {
+ _partitionGroupId = partitionGroupId;
+ _startCheckpoint = startCheckpoint;
+ }
+
+ public void setStartCheckpoint(String startCheckpoint) {
+ _startCheckpoint = startCheckpoint;
+ }
+
+ public int getPartitionGroupId() {
+ return _partitionGroupId;
+ }
+
+ public String getStartCheckpoint() {
+ return _startCheckpoint;
+ }
+}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
index 5b9104e..a9cd2d6 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/stream/StreamMetadataProvider.java
@@ -20,6 +20,7 @@ package org.apache.pinot.spi.stream;
import java.io.Closeable;
import java.util.List;
+import java.util.concurrent.TimeoutException;
import javax.annotation.Nonnull;
import org.apache.pinot.spi.annotations.InterfaceAudience;
import org.apache.pinot.spi.annotations.InterfaceStability;
@@ -39,8 +40,9 @@ public interface StreamMetadataProvider extends Closeable {
int fetchPartitionCount(long timeoutMillis);
// takes the current state of partition groups (groupings of shards, the state of the consumption) and creates the new state
- List<PartitionGroupMetadata> getPartitionGroupMetadataList(
- List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis);
+ List<PartitionGroupInfo> getPartitionGroupMetadataList(
+ List<PartitionGroupMetadata> currentPartitionGroupsMetadata, long timeoutMillis)
+ throws TimeoutException;
// Issue 5953 Retain this interface for 0.5.0, remove in 0.6.0
@Deprecated
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org