You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/01/27 16:30:59 UTC
[PR] Add Kafka ingestion support for subset partitions [pinot]
xiangfu0 opened a new pull request, #17587:
URL: https://github.com/apache/pinot/pull/17587
- pinot-kafka-base: Add stream.kafka.partition.ids config and KafkaPartitionSubsetUtils parser
- pinot-kafka-2.0/3.0: Override fetchPartitionCount, fetchPartitionIds, computePartitionGroupMetadata to respect partition subset; validate configured IDs against topic
- Add unit tests for subset parsing and metadata provider (KafkaPartitionSubsetUtilsTest, KafkaPartitionLevelConsumerTest)
- Add stream subset example (subsetPartitions) and Kafka 2.0 README docs
- QuickStart: Add fineFoodReviews-part-0 and fineFoodReviews-part-1 realtime tables, each consuming one partition of fineFoodReviews topic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2736534887
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -446,13 +478,16 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
instanceToNumPartitionsMap.put(existingInstance, numPartitionsOnInstance + 1);
}
}
+ } else {
+ partitionIdToExistingInstancesMap.add(List.of());
}
Review Comment:
The else clause initializing `partitionIdToExistingInstancesMap` with an empty list should have been added in the original loop structure starting at line 456. This duplicates the pattern from line 470 where existing instances are added conditionally. Consider moving this initialization to ensure the list size matches `partitionIds.size()` consistently throughout the method.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +133,56 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ Optional<List<Integer>> subsetOpt =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subsetOpt.isEmpty()) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = subsetOpt.get();
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
+ LOGGER.warn(
+ "Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ + "Topic partitions: {}",
Review Comment:
The warning message in Kafka 3.0 is less detailed than Kafka 2.0 (lines 157-160 in the 2.0 version). The Kafka 2.0 version includes additional context: "This indicates that topic partitions may have changed between validation and metadata computation. Skipping this partition." This extra explanation should be added to the Kafka 3.0 version for consistency and clarity.
```suggestion
+ "Topic partitions: {}. This indicates that topic partitions may have changed between "
+ "validation and metadata computation. Skipping this partition.",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2874839268
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -400,7 +400,15 @@ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
streamConfigs.forEach(_flushThresholdUpdateManager::clearFlushThresholdUpdater);
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
- int numPartitionGroups = consumeMeta.size();
+ // For subset-partition tables the instance map was built with the full Kafka topic partition
+ // count (set by ImplicitRealtimeTablePartitionSelector), so getNumPartitions() exceeds the
+ // number of actively consumed partitions. Use it so that RealtimeSegmentAssignment correctly
+ // resolves non-contiguous partition IDs via partitionId % numPartitions.
+ // For standard tables the instance map either has a single slot (non-partition-based
+ // assignment) or exactly as many slots as partitions, so consumeMeta.size() is the right value.
+ int numPartitionGroups = instancePartitions.getNumPartitions() > consumeMeta.size()
Review Comment:
This is not correct because instance partitions might not match kafka stream (e.g. when instance assignment is not explicitly configured). We should read the total partitions from the stream, and pass it in.
Same for other places
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2735887132
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +133,52 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ Optional<List<Integer>> subsetOpt =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subsetOpt.isEmpty()) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = subsetOpt.get();
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
Review Comment:
Silently skipping non-existent partitions contradicts the earlier validation. If a partition ID was validated at lines 90-94, it should always exist in topicIds here. This skip suggests partitions could disappear between fetchPartitionIds and this method, which would be a race condition. Either remove this check (trusting the earlier validation) or log a warning if this unexpected state occurs.
```suggestion
if (!topicIds.contains(partitionId)) {
LOGGER.warn(
"Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ "This indicates that topic partitions may have changed between validation and metadata "
+ "computation. Skipping this partition. Current topic partitions: {}",
partitionId, _topic, topicIds);
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +133,52 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ Optional<List<Integer>> subsetOpt =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subsetOpt.isEmpty()) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = subsetOpt.get();
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
Review Comment:
Silently skipping non-existent partitions contradicts the earlier validation. If a partition ID was validated at lines 90-94, it should always exist in topicIds here. This skip suggests partitions could disappear between fetchPartitionIds and this method, which would be a race condition. Either remove this check (trusting the earlier validation) or log a warning if this unexpected state occurs.
```suggestion
if (!topicIds.contains(partitionId)) {
LOGGER.warn("Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ "Topic partitions: {}", partitionId, _topic, topicIds);
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/ImplicitRealtimeTablePartitionSelector.java:
##########
@@ -54,22 +56,35 @@ public ImplicitRealtimeTablePartitionSelector(TableConfig tableConfig,
String tableNameWithType, @Nullable InstancePartitions existingInstancePartitions, boolean minimizeDataMovement,
StreamMetadataProvider streamMetadataProvider) {
super(replicaGroupPartitionConfig, tableNameWithType, existingInstancePartitions, minimizeDataMovement);
- _numPartitions = getStreamNumPartitions(streamMetadataProvider);
- }
-
- private int getStreamNumPartitions(StreamMetadataProvider streamMetadataProvider) {
+ List<Integer> partitionIds = null;
try (streamMetadataProvider) {
- return streamMetadataProvider.fetchPartitionCount(10_000L);
+ _numPartitions = streamMetadataProvider.fetchPartitionCount(10_000L);
+ try {
+ Set<Integer> ids = streamMetadataProvider.fetchPartitionIds(10_000L);
+ if (ids != null && !ids.isEmpty()) {
+ partitionIds = new ArrayList<>(ids);
+ Collections.sort(partitionIds);
+ }
+ } catch (UnsupportedOperationException e) {
+ // Stream does not expose partition IDs; instance partitions will use 0..numPartitions-1
+ }
Review Comment:
Catching and silently ignoring UnsupportedOperationException makes it unclear which streams support fetchPartitionIds and which don't. Consider logging at DEBUG level that the stream doesn't expose partition IDs, so operators can understand why explicit partition IDs weren't used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #17587:
URL: https://github.com/apache/pinot/pull/17587
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2732861641
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/README.md:
##########
@@ -40,3 +40,9 @@ Below is a sample `streamConfigs` used to create a real-time table with Kafka co
"stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
}
```
+
Review Comment:
Corrected incorrect code fence language identifier from '$xslt' to 'json'.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -433,26 +454,33 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
partitionIdToInstanceSetMap.add(instanceSet);
// Keep the existing instances that are still alive
- if (partitionId < existingNumPartitions) {
+ if (idx < existingNumPartitions) {
List<String> existingInstances = _existingInstancePartitions.getInstances(partitionId, replicaGroupId);
- partitionIdToExistingInstancesMap.add(existingInstances);
- int numInstancesToCheck = Math.min(numInstancesPerPartition, existingInstances.size());
- for (int i = 0; i < numInstancesToCheck; i++) {
- String existingInstance = existingInstances.get(i);
- Integer numPartitionsOnInstance = instanceToNumPartitionsMap.get(existingInstance);
- if (numPartitionsOnInstance != null && numPartitionsOnInstance < maxNumPartitionsPerInstance) {
- instances.set(i, existingInstance);
- instanceSet.add(existingInstance);
- instanceToNumPartitionsMap.put(existingInstance, numPartitionsOnInstance + 1);
+ if (existingInstances != null) {
+ partitionIdToExistingInstancesMap.add(existingInstances);
Review Comment:
Multiple locations in this method now check if existingInstances is null and add empty lists. Consider extracting this logic into a helper method to reduce duplication and improve readability.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java:
##########
@@ -160,12 +160,14 @@ protected List<String> assignConsumingSegment(int segmentPartitionId, InstancePa
instancesAssigned.add(instances.get(segmentPartitionId % instances.size()));
}
} else {
- // Explicit partition:
- // Assign segment to the first instance within the partition.
-
+ // Explicit partition: instance partitions are keyed by stream partition id (supports non-contiguous subset).
for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups; replicaGroupId++) {
- int partitionId = segmentPartitionId % numPartitions;
- instancesAssigned.add(instancePartitions.getInstances(partitionId, replicaGroupId).get(0));
+ List<String> instances = instancePartitions.getInstances(segmentPartitionId, replicaGroupId);
+ Preconditions.checkState(instances != null && !instances.isEmpty(),
+ "No instances for partition %s in CONSUMING instance partitions (table: %s). "
+ + "Check stream partition subset config matches instance partition selection.",
Review Comment:
The error message could be more actionable by specifying which configuration properties to check. Consider mentioning 'stream.kafka.partition.ids' explicitly in the error message.
```suggestion
+ "Check that the stream partition subset configuration (for example, 'stream.kafka.partition.ids') "
+ "matches the instance partition selection in the table configuration.",
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -390,24 +404,30 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
}
if (numPartitions == 1) {
+ int partitionId = (getPartitionIds() != null && getPartitionIds().size() == 1)
+ ? getPartitionIds().get(0) : 0;
for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups; replicaGroupId++) {
List<String> instancesInReplicaGroup = replicaGroupIdToInstancesMap.get(replicaGroupId);
if (replicaGroupId < existingNumReplicaGroups) {
- List<String> existingInstances = _existingInstancePartitions.getInstances(0, replicaGroupId);
+ List<String> existingInstances = _existingInstancePartitions.getInstances(partitionId, replicaGroupId);
LinkedHashSet<String> candidateInstances = new LinkedHashSet<>(instancesInReplicaGroup);
List<String> instances =
- selectInstancesWithMinimumMovement(numInstancesPerReplicaGroup, candidateInstances, existingInstances);
+ selectInstancesWithMinimumMovement(numInstancesPerReplicaGroup, candidateInstances,
+ existingInstances != null ? existingInstances : List.of());
Review Comment:
The null check for existingInstances could be simplified by handling it at the source. Consider checking if _existingInstancePartitions.getInstances() returns null before calling selectInstancesWithMinimumMovement to avoid the ternary operator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2760387499
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -82,20 +114,46 @@ public int fetchPartitionCount(long timeoutMillis) {
@Override
public Set<Integer> fetchPartitionIds(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
- }
- Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
- for (PartitionInfo partitionInfo : partitionInfos) {
- partitionIds.add(partitionInfo.partition());
+ if (_partialPartitions) {
+ return new HashSet<>(_partitionIdSubset);
}
- return partitionIds;
+ return toPartitionIdSet(getPartitionInfos(timeoutMillis));
} catch (TimeoutException e) {
throw new TransientConsumerException(e);
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Map<Integer, StreamPartitionMsgOffset> partitionIdToEndOffset =
+ new HashMap<>(partitionGroupConsumptionStatuses.size());
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ partitionIdToEndOffset.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ StreamPartitionMsgOffset startOffset = partitionIdToEndOffset.get(partitionId);
Review Comment:
nit:
```suggestion
StreamPartitionMsgOffset endOffset = partitionIdToEndOffset.get(partitionId);
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -57,23 +63,49 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private final boolean _partialPartitions;
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
+ }
+
+ private List<PartitionInfo> getPartitionInfos(long timeoutMillis) {
+ List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException("Failed to fetch partition information for topic: " + _topic);
+ }
+ return partitionInfos;
+ }
+
+ private Set<Integer> toPartitionIdSet(List<PartitionInfo> partitionInfos) {
+ Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ partitionIds.add(partitionInfo.partition());
+ }
+ return partitionIds;
}
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
+ if (_partialPartitions) {
+ return _partitionIdSubset.size();
Review Comment:
So the partial subset for table will never change?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -82,20 +114,46 @@ public int fetchPartitionCount(long timeoutMillis) {
@Override
public Set<Integer> fetchPartitionIds(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
- }
- Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
- for (PartitionInfo partitionInfo : partitionInfos) {
- partitionIds.add(partitionInfo.partition());
+ if (_partialPartitions) {
+ return new HashSet<>(_partitionIdSubset);
}
- return partitionIds;
+ return toPartitionIdSet(getPartitionInfos(timeoutMillis));
} catch (TimeoutException e) {
throw new TransientConsumerException(e);
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Map<Integer, StreamPartitionMsgOffset> partitionIdToEndOffset =
+ new HashMap<>(partitionGroupConsumptionStatuses.size());
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ partitionIdToEndOffset.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ StreamPartitionMsgOffset startOffset = partitionIdToEndOffset.get(partitionId);
+ if (startOffset == null) {
+ try (StreamMetadataProvider partitionMetadataProvider =
+ streamConsumerFactory.createPartitionMetadataProvider(
+ StreamConsumerFactory.getUniqueClientId(clientId), partitionId)) {
+ startOffset = partitionMetadataProvider.fetchStreamPartitionOffset(
+ streamConfig.getOffsetCriteria(), timeoutMillis);
+ }
+ }
+ result.add(new PartitionGroupMetadata(partitionId, startOffset));
Review Comment:
So `fetchPartitionIds` returns the same subset created in constructor everytime.
But in this method we add new partitionIds, so behaviour seems inconsistent across these methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2851384629
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/segment/RealtimeSegmentAssignment.java:
##########
@@ -160,12 +160,15 @@ protected List<String> assignConsumingSegment(int segmentPartitionId, InstancePa
instancesAssigned.add(instances.get(segmentPartitionId % instances.size()));
}
} else {
- // Explicit partition:
- // Assign segment to the first instance within the partition.
-
+ // Explicit partition: instance partitions are keyed by stream partition id (supports non-contiguous subset).
for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups; replicaGroupId++) {
- int partitionId = segmentPartitionId % numPartitions;
- instancesAssigned.add(instancePartitions.getInstances(partitionId, replicaGroupId).get(0));
+ List<String> instances = instancePartitions.getInstances(segmentPartitionId, replicaGroupId);
Review Comment:
We dont need to do `segmentPartitionId % numPartitions` ?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -343,7 +407,26 @@ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
}
throw new TransientConsumerException(
- new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
+ new RuntimeException("Failed to fetch partition information for topic: " + _topic));
+ }
+
+ private void validatePartitionIds(List<Integer> subset) {
+ List<Integer> topicPartitionIds = new ArrayList<>();
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(10_000L);
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ topicPartitionIds.add(partitionInfo.partition());
+ }
+ Collections.sort(topicPartitionIds);
Review Comment:
nit: No need for this
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -343,7 +407,26 @@ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
}
throw new TransientConsumerException(
- new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
+ new RuntimeException("Failed to fetch partition information for topic: " + _topic));
+ }
+
+ private void validatePartitionIds(List<Integer> subset) {
+ List<Integer> topicPartitionIds = new ArrayList<>();
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(10_000L);
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ topicPartitionIds.add(partitionInfo.partition());
+ }
+ Collections.sort(topicPartitionIds);
+ List<Integer> missingPartitionIds = new ArrayList<>();
+ for (Integer partitionId : subset) {
+ if (!topicPartitionIds.contains(partitionId)) {
Review Comment:
nit: `contains` on list, let use set.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -289,16 +322,18 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
int maxReplicaGroupsPerPool = (numReplicaGroups + numPools - 1) / numPools;
int startIndex = Math.abs(tableNameHash % numPools);
- int existingNumReplicaGroups = _existingInstancePartitions.getNumReplicaGroups();
- int existingNumPartitions = _existingInstancePartitions.getNumPartitions();
+ int existingNumReplicaGroups =
+ _existingInstancePartitions != null ? _existingInstancePartitions.getNumReplicaGroups() : 0;
+ int existingNumPartitions =
+ _existingInstancePartitions != null ? _existingInstancePartitions.getNumPartitions() : 0;
for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups; replicaGroupId++) {
// For each replica-group, gather number of existing instances within each pool
Set<String> existingInstanceSet = new HashSet<>();
replicaGroupIdToExistingInstancesMap.add(existingInstanceSet);
Map<Integer, Integer> poolToNumExistingInstancesMap = new TreeMap<>();
if (replicaGroupId < existingNumReplicaGroups) {
for (int partitionId = 0; partitionId < existingNumPartitions; partitionId++) {
Review Comment:
existingNumPartitions seems to come from `Integer.max(_numPartitions, partitionId + 1);`. This is correct?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/ImplicitRealtimeTablePartitionSelector.java:
##########
@@ -19,30 +19,35 @@
package org.apache.pinot.controller.helix.core.assignment.instance;
import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
import javax.annotation.Nullable;
import org.apache.pinot.common.assignment.InstancePartitions;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.assignment.InstanceReplicaGroupPartitionConfig;
import org.apache.pinot.spi.stream.StreamConsumerFactoryProvider;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.apache.pinot.spi.utils.IngestionConfigUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
- * Variation of {@link InstanceReplicaGroupPartitionSelector} that uses the number of partitions from the stream
- * to determine the number of partitions in each replica group.
+ * Variation of {@link InstanceReplicaGroupPartitionSelector} that uses the number and IDs of partitions
+ * from the stream to determine the partitions in each replica group. When the stream exposes partition IDs
+ * (e.g. Kafka with subset), instance partitions are keyed by those IDs so non-contiguous subsets work.
*/
public class ImplicitRealtimeTablePartitionSelector extends InstanceReplicaGroupPartitionSelector {
+ private static final Logger LOGGER = LoggerFactory.getLogger(ImplicitRealtimeTablePartitionSelector.class);
Review Comment:
seems unused
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2751175185
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -57,23 +63,47 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private final boolean _partialPartitions;
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
}
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
Review Comment:
I think we want to avoid String.format: https://github.com/apache/pinot/issues/14404
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
+ LOGGER.warn(
+ "Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ + "This indicates that topic partitions may have changed between validation and metadata "
+ + "computation. Skipping this partition. Current topic partitions: {}",
+ partitionId, _topic, topicIds);
+ continue;
+ }
+ StreamPartitionMsgOffset startOffset = consumptionByPartition.get(partitionId);
+ if (startOffset == null) {
+ try (StreamMetadataProvider partitionMetadataProvider =
+ streamConsumerFactory.createPartitionMetadataProvider(
+ StreamConsumerFactory.getUniqueClientId(clientId), partitionId)) {
+ startOffset = partitionMetadataProvider.fetchStreamPartitionOffset(
+ streamConfig.getOffsetCriteria(), timeoutMillis);
+ }
+ }
+ result.add(new PartitionGroupMetadata(partitionId, startOffset));
+ }
+ return result;
+ }
+
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis,
+ boolean forceGetOffsetFromStream) throws IOException, java.util.concurrent.TimeoutException {
+ if (forceGetOffsetFromStream) {
+ return computePartitionGroupMetadata(clientId, streamConfig, Collections.emptyList(), timeoutMillis);
+ }
+ return computePartitionGroupMetadata(clientId, streamConfig, partitionGroupConsumptionStatuses, timeoutMillis);
+ }
Review Comment:
Is this method required?
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -57,23 +63,47 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private final boolean _partialPartitions;
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
}
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ }
+ if (_partialPartitions) {
+ Set<Integer> topicIds = new HashSet<>();
+ for (PartitionInfo p : partitionInfos) {
+ topicIds.add(p.partition());
Review Comment:
We are just returning count here, so I think we can skip adding to Set
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -57,23 +63,47 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private final boolean _partialPartitions;
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
}
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
Review Comment:
Applicable for other places as well
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
Review Comment:
```suggestion
Map<Integer, StreamPartitionMsgOffset> partitionIdToEndOffset = new HashMap<>(partitionGroupConsumptionStatuses.size());
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java:
##########
@@ -0,0 +1,79 @@
+/**
+ * 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.plugin.stream.kafka;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+
+/**
+ * Utilities for parsing and validating Kafka partition subset configuration
+ * (stream.kafka.partition.ids) from stream config.
+ */
+public final class KafkaPartitionSubsetUtils {
+
+ private KafkaPartitionSubsetUtils() {
+ }
+
+ /**
+ * Reads the optional comma-separated partition ID list from the stream config map.
+ * Returns a sorted, deduplicated list for stable ordering when used for partition group metadata.
+ * Duplicate IDs in the config are silently removed.
+ *
+ * @param streamConfigMap table stream config map (e.g. from
+ * {@link org.apache.pinot.spi.stream.StreamConfig#getStreamConfigsMap()})
+ * @return Sorted list of unique partition IDs when stream.kafka.partition.ids is set and non-empty;
+ * null when not set or blank
+ * @throws IllegalArgumentException if the value contains invalid (non-integer) entries
+ */
+ @Nullable
+ public static List<Integer> getPartitionIdsFromConfig(Map<String, String> streamConfigMap) {
+ String key = KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS);
+ String value = streamConfigMap.get(key);
+ if (StringUtils.isBlank(value)) {
+ return null;
+ }
+ String[] parts = value.split(",");
+ Set<Integer> idSet = new LinkedHashSet<>(parts.length);
Review Comment:
nit, we are sorting the set anyways so we can just use the simple hashset.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -57,23 +63,47 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ private final boolean _partialPartitions;
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
}
@Override
public int fetchPartitionCount(long timeoutMillis) {
try {
List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isNotEmpty(partitionInfos)) {
- return partitionInfos.size();
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
+ }
+ if (_partialPartitions) {
+ Set<Integer> topicIds = new HashSet<>();
+ for (PartitionInfo p : partitionInfos) {
+ topicIds.add(p.partition());
Review Comment:
Same comments of kafka20 class
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
Review Comment:
```suggestion
Set<Integer> partitionIds = fetchPartitionIds(timeoutMillis);
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -86,6 +116,19 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
if (CollectionUtils.isEmpty(partitionInfos)) {
throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
}
+ if (_partialPartitions) {
+ Set<Integer> topicIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
Review Comment:
nit
```suggestion
Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2742499282
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
+ LOGGER.warn(
+ "Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ + "This indicates that topic partitions may have changed between validation and metadata "
+ + "computation. Skipping this partition. Current topic partitions: {}",
+ partitionId, _topic, topicIds);
Review Comment:
The logging message splits arguments across multiple lines in an inconsistent order. The message template references 3 placeholders but lists topicIds last, while the previous validation error at line 101 lists it third. For consistency, move 'Current topic partitions: {}' to position 3 in both the message and arguments, matching the error message pattern.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
+ LOGGER.warn(
+ "Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ + "Current topic partitions: {}. This indicates that topic partitions may have changed between "
+ + "validation and metadata computation. Skipping this partition.",
Review Comment:
The warn message structure differs from the Kafka 2.0 implementation. The Kafka 2.0 version (line 162-163) places the explanatory text about partition changes before 'Current topic partitions', while this version places it after. Standardize the message order between Kafka 2.0 and 3.0 implementations for consistency.
```suggestion
+ "This indicates that topic partitions may have changed between validation and metadata computation. "
+ "Current topic partitions: {}. Skipping this partition.",
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2741686543
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -390,24 +416,29 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
}
if (numPartitions == 1) {
+ int partitionId = (getPartitionIds() != null && getPartitionIds().size() == 1)
+ ? getPartitionIds().get(0) : 0;
for (int replicaGroupId = 0; replicaGroupId < numReplicaGroups; replicaGroupId++) {
List<String> instancesInReplicaGroup = replicaGroupIdToInstancesMap.get(replicaGroupId);
if (replicaGroupId < existingNumReplicaGroups) {
- List<String> existingInstances = _existingInstancePartitions.getInstances(0, replicaGroupId);
+ List<String> existingInstances = getExistingInstancesOrEmpty(partitionId, replicaGroupId);
LinkedHashSet<String> candidateInstances = new LinkedHashSet<>(instancesInReplicaGroup);
List<String> instances =
selectInstancesWithMinimumMovement(numInstancesPerReplicaGroup, candidateInstances, existingInstances);
LOGGER.info(
- "Selecting instances: {} for replica-group: {}, partition: 0 for table: {}, existing instances: {}",
- instances, replicaGroupId, _tableNameWithType, existingInstances);
- instancePartitions.setInstances(0, replicaGroupId, instances);
+ "Selecting instances: {} for replica-group: {}, partition: {} for table: {}, existing instances: {}",
+ instances, replicaGroupId, partitionId, _tableNameWithType, existingInstances);
+ instancePartitions.setInstances(partitionId, replicaGroupId, instances);
} else {
- LOGGER.info("Selecting instances: {} for replica-group: {}, partition: 0 for table: {}, "
- + "there is no existing instances", instancesInReplicaGroup, replicaGroupId, _tableNameWithType);
- instancePartitions.setInstances(0, replicaGroupId, instancesInReplicaGroup);
+ LOGGER.info("Selecting instances: {} for replica-group: {}, partition: {} for table: {}, "
+ + "there is no existing instances", instancesInReplicaGroup, replicaGroupId, partitionId,
+ _tableNameWithType);
+ instancePartitions.setInstances(partitionId, replicaGroupId, instancesInReplicaGroup);
}
}
} else {
+ List<Integer> partitionIds = getPartitionIds() != null ? getPartitionIds()
+ : IntStream.range(0, numPartitions).boxed().collect(Collectors.toList());
Review Comment:
The method getPartitionIds() is called twice. Cache the result in a local variable to avoid the redundant method call.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -196,10 +210,12 @@ private void replicaGroupBasedSimple(Map<Integer, List<InstanceConfig>> poolToIn
// [i0, i1, i2, i3, i4]
// p0 p0 p0 p1 p1
// p1 p2 p2 p2
+ List<Integer> partitionIds = getPartitionIds() != null ? getPartitionIds()
+ : IntStream.range(0, numPartitions).boxed().collect(Collectors.toList());
Review Comment:
The method getPartitionIds() is called twice. Cache the result in a local variable to avoid the redundant method call.
```suggestion
List<Integer> partitionIds = getPartitionIds();
if (partitionIds == null) {
partitionIds = IntStream.range(0, numPartitions).boxed().collect(Collectors.toList());
}
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -96,6 +139,55 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Set<Integer> topicIds = fetchPartitionIds(timeoutMillis);
+ Map<Integer, StreamPartitionMsgOffset> consumptionByPartition = new HashMap<>();
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ consumptionByPartition.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ if (!topicIds.contains(partitionId)) {
+ LOGGER.warn(
+ "Configured partition id {} does not exist in topic {} when computing partition group metadata. "
+ + "Topic partitions: {}. This indicates that topic partitions may have changed between "
Review Comment:
The warning message spans multiple lines with inconsistent formatting. The second line is missing 'Current' prefix compared to the kafka-2.0 implementation. Consider using a consistent format between both implementations or using a format string placeholder pattern for better maintainability.
```suggestion
+ "Current topic partitions: {}. This indicates that topic partitions may have changed between "
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -390,24 +416,29 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
}
if (numPartitions == 1) {
+ int partitionId = (getPartitionIds() != null && getPartitionIds().size() == 1)
+ ? getPartitionIds().get(0) : 0;
Review Comment:
The method getPartitionIds() is called three times. Cache the result in a local variable to avoid redundant method calls.
```suggestion
List<Integer> partitionIds = getPartitionIds();
int partitionId = (partitionIds != null && partitionIds.size() == 1)
? partitionIds.get(0) : 0;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "noob-se7en (via GitHub)" <gi...@apache.org>.
noob-se7en commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2760400439
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -82,20 +114,46 @@ public int fetchPartitionCount(long timeoutMillis) {
@Override
public Set<Integer> fetchPartitionIds(long timeoutMillis) {
try {
- List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
- if (CollectionUtils.isEmpty(partitionInfos)) {
- throw new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic));
- }
- Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
- for (PartitionInfo partitionInfo : partitionInfos) {
- partitionIds.add(partitionInfo.partition());
+ if (_partialPartitions) {
+ return new HashSet<>(_partitionIdSubset);
}
- return partitionIds;
+ return toPartitionIdSet(getPartitionInfos(timeoutMillis));
} catch (TimeoutException e) {
throw new TransientConsumerException(e);
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
+ if (!_partialPartitions) {
+ return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,
+ partitionGroupConsumptionStatuses, timeoutMillis);
+ }
+ List<Integer> subset = _partitionIdSubset;
+ Map<Integer, StreamPartitionMsgOffset> partitionIdToEndOffset =
+ new HashMap<>(partitionGroupConsumptionStatuses.size());
+ for (PartitionGroupConsumptionStatus s : partitionGroupConsumptionStatuses) {
+ partitionIdToEndOffset.put(s.getStreamPartitionGroupId(), s.getEndOffset());
+ }
+ StreamConsumerFactory streamConsumerFactory = StreamConsumerFactoryProvider.create(streamConfig);
+ List<PartitionGroupMetadata> result = new ArrayList<>(subset.size());
+ for (Integer partitionId : subset) {
+ StreamPartitionMsgOffset startOffset = partitionIdToEndOffset.get(partitionId);
+ if (startOffset == null) {
+ try (StreamMetadataProvider partitionMetadataProvider =
+ streamConsumerFactory.createPartitionMetadataProvider(
+ StreamConsumerFactory.getUniqueClientId(clientId), partitionId)) {
+ startOffset = partitionMetadataProvider.fetchStreamPartitionOffset(
+ streamConfig.getOffsetCriteria(), timeoutMillis);
+ }
+ }
+ result.add(new PartitionGroupMetadata(partitionId, startOffset));
Review Comment:
So `fetchPartitionIds` returns the same subset created in constructor everytime.
But in this method we add new partitionIds, so behaviour seems inconsistent across these methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2901507866
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -58,13 +64,26 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ /**
+ * Immutable partition ID subset from table config. Read once at construction; does not change during the
+ * provider's lifetime. Empty when no subset is configured (consume all partitions).
+ */
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ validatePartitionIds(_partitionIdSubset);
+ } else {
+ _partitionIdSubset = Collections.emptyList();
Review Comment:
(nit) `List.of()`
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -397,12 +397,14 @@ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
String realtimeTableName = tableConfig.getTableName();
LOGGER.info("Setting up new LLC table: {}", realtimeTableName);
- int numPartitionGroups = 0;
+ List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
+ int numActivePartitionGroups = 0;
for (StreamMetadata streamMetadata : streamMetadataList) {
_flushThresholdUpdateManager.clearFlushThresholdUpdater(streamMetadata.getStreamConfig());
- numPartitionGroups += streamMetadata.getNumPartitions();
+ numActivePartitionGroups += streamMetadata.getPartitionGroupMetadataList().size();
}
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
+ int numPartitionGroups = getPartitionCountForRouting(streamConfigs, numActivePartitionGroups);
Review Comment:
This info should be available from `StreamMetadata`. No need to access the remote stream
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -99,6 +118,40 @@ public Set<Integer> fetchPartitionIds(long timeoutMillis) {
}
}
+ @Override
+ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientId, StreamConfig streamConfig,
+ List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
+ throws IOException, java.util.concurrent.TimeoutException {
Review Comment:
(minor) Import `TimeoutException`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2922734539
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2123,6 +2125,38 @@ private int getNumReplicas(TableConfig tableConfig, InstancePartitions instanceP
}
}
+ /**
+ * Gets the total partition count for routing segment assignment.
+ *
+ * <p>For subset-partition tables, this MUST return the total Kafka topic partition count (not the
+ * subset size) so that RealtimeSegmentAssignment correctly routes non-contiguous partition IDs via
+ * {@code partitionId % totalPartitions}.
+ *
+ * <p>For standard tables, this returns the number of actively consumed partitions.
+ *
+ * <p>For tables with multiple streams, this sums the partition counts across all streams.
+ *
+ * @param streamMetadataList the list of stream metadata (already fetched, no remote calls needed)
+ * @param activePartitionCount the number of partitions being actively consumed (from partition group metadata)
+ * @return the partition count to use for segment assignment routing
+ */
+ private int getPartitionCountForRouting(List<StreamMetadata> streamMetadataList, int activePartitionCount) {
+ int totalPartitionCount = 0;
+ for (StreamMetadata streamMetadata : streamMetadataList) {
+ totalPartitionCount += streamMetadata.getNumPartitions();
+ }
+
+ // If total partition count > active count, this is a subset-partition table
+ // Use total count for proper modulo routing
+ if (totalPartitionCount > activePartitionCount) {
Review Comment:
I still don't follow why do we need to compare them. What is the problem of always using totalPartitionCount?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2922921799
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -2123,6 +2125,38 @@ private int getNumReplicas(TableConfig tableConfig, InstancePartitions instanceP
}
}
+ /**
+ * Gets the total partition count for routing segment assignment.
+ *
+ * <p>For subset-partition tables, this MUST return the total Kafka topic partition count (not the
+ * subset size) so that RealtimeSegmentAssignment correctly routes non-contiguous partition IDs via
+ * {@code partitionId % totalPartitions}.
+ *
+ * <p>For standard tables, this returns the number of actively consumed partitions.
+ *
+ * <p>For tables with multiple streams, this sums the partition counts across all streams.
+ *
+ * @param streamMetadataList the list of stream metadata (already fetched, no remote calls needed)
+ * @param activePartitionCount the number of partitions being actively consumed (from partition group metadata)
+ * @return the partition count to use for segment assignment routing
+ */
+ private int getPartitionCountForRouting(List<StreamMetadata> streamMetadataList, int activePartitionCount) {
+ int totalPartitionCount = 0;
+ for (StreamMetadata streamMetadata : streamMetadataList) {
+ totalPartitionCount += streamMetadata.getNumPartitions();
+ }
+
+ // If total partition count > active count, this is a subset-partition table
+ // Use total count for proper modulo routing
+ if (totalPartitionCount > activePartitionCount) {
Review Comment:
You're right — there's no good reason for the comparison. We can just always return totalPartitionCount.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2848682916
##########
pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java:
##########
@@ -1023,4 +1042,67 @@ private static String firstNonEmpty(String... values) {
}
return null;
}
+
+ protected void createFineFoodReviewsFederatedTable() {
+ if (!useDefaultBootstrapTableDir()) {
+ return;
+ }
+ Map<String, String> streamTableDirectories = getDefaultStreamTableDirectories();
+ if (!streamTableDirectories.containsKey("fineFoodReviews-part-0")
+ || !streamTableDirectories.containsKey("fineFoodReviews-part-1")) {
+ return;
+ }
+ String logicalTableName = "fineFoodReviews-federated";
+ try {
+ Schema schema = loadSchemaFromResource("/examples/stream/fineFoodReviews/fineFoodReviews_schema.json");
+ schema.setSchemaName(logicalTableName);
+ createSchemaOnController(schema, logicalTableName);
+
+ LogicalTableConfig logicalTableConfig = new LogicalTableConfig();
+ logicalTableConfig.setTableName(logicalTableName);
+ logicalTableConfig.setBrokerTenant("DefaultTenant");
+ logicalTableConfig.setRefRealtimeTableName("fineFoodReviews-part-0_REALTIME");
+ logicalTableConfig.setPhysicalTableConfigMap(Map.of(
+ "fineFoodReviews-part-0_REALTIME", new PhysicalTableConfig(),
+ "fineFoodReviews-part-1_REALTIME", new PhysicalTableConfig()
+ ));
+
+ String logicalTableUrl = "http://localhost:" + QuickstartRunner.DEFAULT_CONTROLLER_PORT + "/logicalTables";
+ AbstractBaseAdminCommand.sendPostRequest(logicalTableUrl, logicalTableConfig.toSingleLineJsonString());
+ printStatus(Quickstart.Color.GREEN,
+ "***** Logical table fineFoodReviews-federated created successfully *****");
+ } catch (Exception e) {
Review Comment:
This `catch (Exception)` only prints `e.getMessage()` and then continues, which makes quickstart failures hard to debug (stack trace and root cause are lost). Consider logging the full exception (or at least `e` + stack trace) via the existing logger so failures can be diagnosed.
```suggestion
} catch (Exception e) {
LOGGER.warn("Logical table {} creation failed: {}", logicalTableName, e.getMessage());
LOGGER.debug("Exception while creating logical table {}", logicalTableName, e);
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionLevelStreamConfig.java:
##########
@@ -120,6 +120,10 @@ public boolean isPopulateMetadata() {
return _populateMetadata;
}
+ public Map<String, String> getStreamConfigMap() {
+ return _streamConfigMap;
+ }
Review Comment:
`getStreamConfigMap()` exposes the internal `_streamConfigMap` by reference, so callers could mutate it and implicitly change consumer behavior after construction. Consider returning an unmodifiable view (or a defensive copy) to keep the stream config immutable once parsed.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java:
##########
@@ -344,6 +408,25 @@ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
}
+ private void validatePartitionIds(List<Integer> subset) {
+ List<Integer> topicPartitionIds = new ArrayList<>();
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(10_000L);
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ topicPartitionIds.add(partitionInfo.partition());
+ }
+ Collections.sort(topicPartitionIds);
+ List<Integer> missingPartitionIds = new ArrayList<>();
+ for (Integer partitionId : subset) {
+ if (!topicPartitionIds.contains(partitionId)) {
Review Comment:
`validatePartitionIds()` does `topicPartitionIds.contains(partitionId)` inside a loop, making validation O(n*m) for large topics. Consider building a `Set` for membership checks (while keeping the sorted list only for the error message) to avoid quadratic behavior.
```suggestion
Set<Integer> topicPartitionIdSet = new HashSet<>(topicPartitionIds);
List<Integer> missingPartitionIds = new ArrayList<>();
for (Integer partitionId : subset) {
if (!topicPartitionIdSet.contains(partitionId)) {
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -346,6 +410,25 @@ private List<PartitionInfo> fetchPartitionInfos(long timeoutMillis) {
new RuntimeException(String.format("Failed to fetch partition information for topic: %s", _topic)));
}
+ private void validatePartitionIds(List<Integer> subset) {
+ List<Integer> topicPartitionIds = new ArrayList<>();
+ List<PartitionInfo> partitionInfos = fetchPartitionInfos(10_000L);
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ topicPartitionIds.add(partitionInfo.partition());
+ }
+ Collections.sort(topicPartitionIds);
+ List<Integer> missingPartitionIds = new ArrayList<>();
+ for (Integer partitionId : subset) {
+ if (!topicPartitionIds.contains(partitionId)) {
Review Comment:
`validatePartitionIds()` does `topicPartitionIds.contains(partitionId)` inside a loop, making validation O(n*m) for large topics. Consider building a `Set` for membership checks (while keeping the sorted list only for the error message) to avoid quadratic behavior.
```suggestion
Set<Integer> topicPartitionIdSet = new HashSet<>(topicPartitionIds);
List<Integer> missingPartitionIds = new ArrayList<>();
for (Integer partitionId : subset) {
if (!topicPartitionIdSet.contains(partitionId)) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2880059159
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -400,7 +400,7 @@ public void setUpNewTable(TableConfig tableConfig, IdealState idealState,
List<StreamConfig> streamConfigs = IngestionConfigUtils.getStreamConfigs(tableConfig);
streamConfigs.forEach(_flushThresholdUpdateManager::clearFlushThresholdUpdater);
InstancePartitions instancePartitions = getConsumingInstancePartitions(tableConfig);
- int numPartitionGroups = consumeMeta.size();
+ int numPartitionGroups = getPartitionCountForRouting(streamConfigs, consumeMeta.size());
Review Comment:
Can you calculate this on the caller side and pass it in? This info should be available when calculating the `consumeMeta`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2841534901
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java:
##########
@@ -0,0 +1,79 @@
+/**
+ * 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.plugin.stream.kafka;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+
+/**
+ * Utilities for parsing and validating Kafka partition subset configuration
+ * (stream.kafka.partition.ids) from stream config.
+ */
+public final class KafkaPartitionSubsetUtils {
+
+ private KafkaPartitionSubsetUtils() {
+ }
+
+ /**
+ * Reads the optional comma-separated partition ID list from the stream config map.
+ * Returns a sorted, deduplicated list for stable ordering when used for partition group metadata.
+ * Duplicate IDs in the config are silently removed to avoid duplicate processing of the same partition.
+ *
+ * @param streamConfigMap table stream config map (e.g. from
+ * {@link org.apache.pinot.spi.stream.StreamConfig#getStreamConfigsMap()})
+ * @return Sorted list of unique partition IDs when stream.kafka.partition.ids is set and non-empty;
+ * null when not set or blank
+ * @throws IllegalArgumentException if the value contains invalid (non-integer) entries
+ */
+ @Nullable
+ public static List<Integer> getPartitionIdsFromConfig(Map<String, String> streamConfigMap) {
+ String key = KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS);
+ String value = streamConfigMap.get(key);
+ if (StringUtils.isBlank(value)) {
+ return null;
+ }
+ String[] parts = value.split(",");
+ Set<Integer> idSet = new HashSet<>(parts.length);
+ for (String part : parts) {
+ String trimmed = part.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ try {
+ idSet.add(Integer.parseInt(trimmed));
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: expected comma-separated integers, got '" + value + "'", e);
+ }
+ }
+ if (idSet.isEmpty()) {
+ return null;
+ }
+ List<Integer> ids = new ArrayList<>(idSet);
+ Collections.sort(ids);
+ return ids;
+ }
Review Comment:
The method accepts negative partition IDs without validation. Kafka partition IDs are always non-negative integers. Consider adding validation to reject negative values and throw an IllegalArgumentException with a clear message, such as "Partition IDs must be non-negative". This will provide early feedback if a user mistakenly configures negative values like "-1,0,1".
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java:
##########
@@ -58,18 +64,54 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ /** Whether this table consumes only a subset of topic partitions (from stream.kafka.partition.ids). */
+ private final boolean _partialPartitions;
+ /**
+ * Immutable partition ID subset from table config. Read once at construction; does not change during the
+ * provider's lifetime. To change the subset, update the table config and restart the consumer.
+ */
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ // The partition subset comes from the table config and is expected to remain stable until config update.
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
+ }
Review Comment:
The PR description mentions "validate configured IDs against topic metadata", but the implementation doesn't validate that the configured partition IDs actually exist in the Kafka topic. While implicit validation will occur when attempting to consume (resulting in runtime errors), early validation during metadata provider construction could provide clearer feedback. Consider adding validation that fetches topic partition metadata and verifies all configured partition IDs exist, or update the PR description if this validation was intentionally deferred to runtime.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -58,18 +64,54 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ /** Whether this table consumes only a subset of topic partitions (from stream.kafka.partition.ids). */
+ private final boolean _partialPartitions;
+ /**
+ * Immutable partition ID subset from table config. Read once at construction; does not change during the
+ * provider's lifetime. To change the subset, update the table config and restart the consumer.
+ */
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ // The partition subset comes from the table config and is expected to remain stable until config update.
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
+ }
Review Comment:
The PR description mentions "validate configured IDs against topic metadata", but the implementation doesn't validate that the configured partition IDs actually exist in the Kafka topic. While implicit validation will occur when attempting to consume (resulting in runtime errors), early validation during metadata provider construction could provide clearer feedback. Consider adding validation that fetches topic partition metadata and verifies all configured partition IDs exist, or update the PR description if this validation was intentionally deferred to runtime.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -58,18 +64,54 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ /** Whether this table consumes only a subset of topic partitions (from stream.kafka.partition.ids). */
+ private final boolean _partialPartitions;
+ /**
+ * Immutable partition ID subset from table config. Read once at construction; does not change during the
+ * provider's lifetime. To change the subset, update the table config and restart the consumer.
+ */
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ // The partition subset comes from the table config and is expected to remain stable until config update.
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
+ }
+
+ private List<PartitionInfo> getPartitionInfos(long timeoutMillis) {
+ List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException("Failed to fetch partition information for topic: " + _topic);
+ }
+ return partitionInfos;
+ }
+
+ private Set<Integer> toPartitionIdSet(List<PartitionInfo> partitionInfos) {
+ Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ partitionIds.add(partitionInfo.partition());
+ }
+ return partitionIds;
}
Review Comment:
These two private methods (`getPartitionInfos` and `toPartitionIdSet`) are defined but never used in this class. The `fetchPartitionCount` and `fetchPartitionIds` methods call `fetchPartitionInfos` directly instead of using these helpers. Consider removing these unused methods to reduce code clutter, or if they were intended for future use, document why they're being kept.
```suggestion
```
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java:
##########
@@ -58,18 +64,54 @@ public class KafkaStreamMetadataProvider extends KafkaPartitionLevelConnectionHa
implements StreamMetadataProvider {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaStreamMetadataProvider.class);
+ /** Whether this table consumes only a subset of topic partitions (from stream.kafka.partition.ids). */
+ private final boolean _partialPartitions;
+ /**
+ * Immutable partition ID subset from table config. Read once at construction; does not change during the
+ * provider's lifetime. To change the subset, update the table config and restart the consumer.
+ */
+ private final List<Integer> _partitionIdSubset;
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig) {
this(clientId, streamConfig, Integer.MIN_VALUE);
}
public KafkaStreamMetadataProvider(String clientId, StreamConfig streamConfig, int partition) {
super(clientId, streamConfig, partition);
+ List<Integer> subset =
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(_config.getStreamConfigMap());
+ if (subset != null) {
+ // The partition subset comes from the table config and is expected to remain stable until config update.
+ _partialPartitions = true;
+ _partitionIdSubset = Collections.unmodifiableList(subset);
+ } else {
+ _partialPartitions = false;
+ _partitionIdSubset = Collections.emptyList();
+ }
+ }
+
+ private List<PartitionInfo> getPartitionInfos(long timeoutMillis) {
+ List<PartitionInfo> partitionInfos = _consumer.partitionsFor(_topic, Duration.ofMillis(timeoutMillis));
+ if (CollectionUtils.isEmpty(partitionInfos)) {
+ throw new RuntimeException("Failed to fetch partition information for topic: " + _topic);
+ }
+ return partitionInfos;
+ }
+
+ private Set<Integer> toPartitionIdSet(List<PartitionInfo> partitionInfos) {
+ Set<Integer> partitionIds = Sets.newHashSetWithExpectedSize(partitionInfos.size());
+ for (PartitionInfo partitionInfo : partitionInfos) {
+ partitionIds.add(partitionInfo.partition());
+ }
+ return partitionIds;
}
Review Comment:
These two private methods (`getPartitionInfos` and `toPartitionIdSet`) are defined but never used in this class. The `fetchPartitionCount` and `fetchPartitionIds` methods call `fetchPartitionInfos` directly instead of using these helpers. Consider removing these unused methods to reduce code clutter, or if they were intended for future use, document why they're being kept.
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2851702767
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/logicaltable/LogicalTableWithTwoRealtimeTableIntegrationTest.java:
##########
@@ -0,0 +1,251 @@
+/**
+ * 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.integration.tests.logicaltable;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.primitives.Longs;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.pinot.plugin.inputformat.avro.AvroUtils;
+import org.apache.pinot.spi.config.table.QueryConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
+import org.apache.pinot.spi.config.table.ingestion.StreamIngestionConfig;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.apache.pinot.util.TestUtils;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class LogicalTableWithTwoRealtimeTableIntegrationTest extends BaseLogicalTableIntegrationTest {
Review Comment:
Shall we add an integration test that does some chaos and random testing
Setup
1. create 1 topic with N partitions
2. create 1 control table which has default configs (consumes from all partitions)
3. create M tables which have some random assignment (non contiguous) of these N partitions. Put these M tables behind 1 logical table.
Random situations
1. Server restart / controller restart
2. Force commit or pause / resume ingestion
3. Anything else that could affect partition level state
The primary need for such a test would be to help prevent future regressions where some functionality depends on getNumPartitions() instead of getPartitionIds()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17587:
URL: https://github.com/apache/pinot/pull/17587#discussion_r2768632079
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/test/java/org/apache/pinot/plugin/stream/kafka20/KafkaPartitionLevelConsumerTest.java:
##########
@@ -202,6 +202,86 @@ public void testGetPartitionCount() {
assertEquals(streamMetadataProvider.fetchPartitionCount(10000L), 2);
}
+ @Test
+ public void testFetchPartitionCountWithSubset()
+ throws Exception {
+ String streamType = "kafka";
+ String streamKafkaBrokerList = _kafkaBrokerAddress;
+ String clientId = "clientId";
+ String tableNameWithType = "tableName_REALTIME";
+
+ // TEST_TOPIC_2 has 2 partitions (0, 1). Subset "0" -> count 1.
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", streamType);
+ streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_2);
+ streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+ streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+ streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+ streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0");
+ StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+ try (KafkaStreamMetadataProvider provider = new KafkaStreamMetadataProvider(clientId, streamConfig)) {
+ Assert.assertEquals(provider.fetchPartitionCount(10000L), 1);
+ }
+
+ // Subset "0,1" -> count 2.
+ streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0,1");
+ streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+ try (KafkaStreamMetadataProvider provider = new KafkaStreamMetadataProvider(clientId, streamConfig)) {
+ Assert.assertEquals(provider.fetchPartitionCount(10000L), 2);
+ }
+ }
+
+ @Test
+ public void testFetchPartitionIdsWithSubset()
+ throws Exception {
+ String streamType = "kafka";
+ String streamKafkaBrokerList = _kafkaBrokerAddress;
+ String clientId = "clientId";
+ String tableNameWithType = "tableName_REALTIME";
+
+ Map<String, String> streamConfigMap = new HashMap<>();
+ streamConfigMap.put("streamType", streamType);
+ streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_2);
+ streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+ streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+ streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+ streamConfigMap.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "1,0");
+ StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+ try (KafkaStreamMetadataProvider provider = new KafkaStreamMetadataProvider(clientId, streamConfig)) {
+ Set<Integer> ids = provider.fetchPartitionIds(10000L);
+ Assert.assertEquals(ids, Set.of(0, 1));
+ }
+ }
+
+ @Test
+ public void testPartitionSubsetDoesNotValidateAtRuntime()
Review Comment:
The test name `testPartitionSubsetDoesNotValidateAtRuntime` is misleading. The test demonstrates that invalid partition IDs (partition 99 doesn't exist) are accepted without validation. Consider renaming to `testPartitionSubsetAcceptsInvalidPartitionIds` or `testPartitionSubsetSkipsValidation` to more accurately describe the behavior being tested.
```suggestion
public void testPartitionSubsetAcceptsInvalidPartitionIds()
```
##########
pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java:
##########
@@ -968,4 +987,66 @@ protected void runVectorQueryExamples(QuickstartRunner runner)
printStatus(Quickstart.Color.YELLOW, prettyPrintResponse(runner.runQuery(q7)));
printStatus(Quickstart.Color.GREEN, "***************************************************");
}
+
+ protected void createFineFoodReviewsFederatedTable() {
+ if (!useDefaultBootstrapTableDir()) {
+ return;
+ }
+ Map<String, String> streamTableDirectories = getDefaultStreamTableDirectories();
+ if (!streamTableDirectories.containsKey("fineFoodReviews-part-0")
+ || !streamTableDirectories.containsKey("fineFoodReviews-part-1")) {
+ return;
+ }
+ String logicalTableName = "fineFoodReviews-federated";
+ try {
+ Schema schema = loadSchemaFromResource("/examples/stream/fineFoodReviews/fineFoodReviews_schema.json");
+ schema.setSchemaName(logicalTableName);
+ createSchemaOnController(schema, logicalTableName);
+
+ LogicalTableConfig logicalTableConfig = new LogicalTableConfig();
+ logicalTableConfig.setTableName(logicalTableName);
+ logicalTableConfig.setBrokerTenant("DefaultTenant");
+ logicalTableConfig.setRefRealtimeTableName("fineFoodReviews-part-0_REALTIME");
+ logicalTableConfig.setPhysicalTableConfigMap(Map.of(
+ "fineFoodReviews-part-0_REALTIME", new PhysicalTableConfig(),
+ "fineFoodReviews-part-1_REALTIME", new PhysicalTableConfig()
+ ));
+
+ String logicalTableUrl = "http://localhost:" + QuickstartRunner.DEFAULT_CONTROLLER_PORT + "/logicalTables";
+ AbstractBaseAdminCommand.sendPostRequest(logicalTableUrl, logicalTableConfig.toSingleLineJsonString());
+ printStatus(Quickstart.Color.GREEN,
+ "***** Logical table fineFoodReviews-federated created successfully *****");
+ } catch (Exception e) {
+ printStatus(Quickstart.Color.YELLOW,
+ "***** Logical table fineFoodReviews-federated creation failed: " + e.getMessage() + " *****");
+ }
+ }
+
+ private Schema loadSchemaFromResource(String resourcePath)
+ throws IOException {
+ try (InputStream inputStream = getClass().getResourceAsStream(resourcePath)) {
+ if (inputStream == null) {
+ throw new IOException("Schema file not found: " + resourcePath);
+ }
+ String schemaJsonString = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
+ return Schema.fromString(schemaJsonString);
+ }
+ }
+
+ private void createSchemaOnController(Schema schema, String logicalTableName)
+ throws Exception {
+ File tempSchemaFile = File.createTempFile(logicalTableName + "_schema", ".json");
+ tempSchemaFile.deleteOnExit();
+ FileUtils.writeStringToFile(tempSchemaFile, schema.toSingleLineJsonString(), StandardCharsets.UTF_8);
+ HttpEntity multipartEntity = MultipartEntityBuilder.create()
+ .addPart("schema", new FileBody(tempSchemaFile, ContentType.APPLICATION_JSON, logicalTableName + ".json"))
+ .build();
+ HttpClient httpClient = new HttpClient();
+ String schemaUrl = "http://localhost:" + QuickstartRunner.DEFAULT_CONTROLLER_PORT + "/schemas?override=true"
+ + "&force=true";
+ SimpleHttpResponse response = httpClient.sendPostRequest(URI.create(schemaUrl), multipartEntity, null, null);
+ if (response.getStatusCode() != 200) {
+ throw new RuntimeException("Schema creation response: " + response.getResponse());
Review Comment:
The HttpClient instance should be closed after use to release resources properly. Consider using try-with-resources if HttpClient implements AutoCloseable, or ensure close() is called in a finally block.
```suggestion
try (HttpClient httpClient = new HttpClient()) {
String schemaUrl = "http://localhost:" + QuickstartRunner.DEFAULT_CONTROLLER_PORT + "/schemas?override=true"
+ "&force=true";
SimpleHttpResponse response = httpClient.sendPostRequest(URI.create(schemaUrl), multipartEntity, null, null);
if (response.getStatusCode() != 200) {
throw new RuntimeException("Schema creation response: " + response.getResponse());
}
```
##########
pinot-tools/src/main/java/org/apache/pinot/tools/QuickStartBase.java:
##########
@@ -81,17 +94,19 @@ public abstract class QuickStartBase {
"examples/batch/testUnnest",
};
- protected static final Map<String, String> DEFAULT_STREAM_TABLE_DIRECTORIES = Map.of(
- "airlineStats", "examples/stream/airlineStats",
- "dailySales", "examples/stream/dailySales",
- "githubEvents", "examples/stream/githubEvents",
- "meetupRsvp", "examples/stream/meetupRsvp",
- "meetupRsvpJson", "examples/stream/meetupRsvpJson",
- "meetupRsvpComplexType", "examples/stream/meetupRsvpComplexType",
- "upsertMeetupRsvp", "examples/stream/upsertMeetupRsvp",
- "upsertJsonMeetupRsvp", "examples/stream/upsertJsonMeetupRsvp",
- "upsertPartialMeetupRsvp", "examples/stream/upsertPartialMeetupRsvp",
- "fineFoodReviews", "examples/stream/fineFoodReviews");
+ protected static final Map<String, String> DEFAULT_STREAM_TABLE_DIRECTORIES = Map.ofEntries(
Review Comment:
The change from `Map.of()` to `Map.ofEntries()` is unnecessary for 10 entries. `Map.of()` supports up to 10 key-value pairs and is more concise. `Map.ofEntries()` is typically used when you have more than 10 entries or need to construct entries programmatically. Consider reverting to `Map.of()` for better readability.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java:
##########
@@ -0,0 +1,79 @@
+/**
+ * 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.plugin.stream.kafka;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+
+
+/**
+ * Utilities for parsing and validating Kafka partition subset configuration
+ * (stream.kafka.partition.ids) from stream config.
+ */
+public final class KafkaPartitionSubsetUtils {
+
+ private KafkaPartitionSubsetUtils() {
+ }
+
+ /**
+ * Reads the optional comma-separated partition ID list from the stream config map.
+ * Returns a sorted, deduplicated list for stable ordering when used for partition group metadata.
+ * Duplicate IDs in the config are silently removed.
+ *
+ * @param streamConfigMap table stream config map (e.g. from
+ * {@link org.apache.pinot.spi.stream.StreamConfig#getStreamConfigsMap()})
+ * @return Sorted list of unique partition IDs when stream.kafka.partition.ids is set and non-empty;
+ * null when not set or blank
+ * @throws IllegalArgumentException if the value contains invalid (non-integer) entries
+ */
+ @Nullable
+ public static List<Integer> getPartitionIdsFromConfig(Map<String, String> streamConfigMap) {
Review Comment:
The method documentation states that duplicate IDs are "silently removed" but doesn't explain why this is the desired behavior. Consider adding a sentence explaining the rationale (e.g., "This ensures stable ordering and prevents duplicate processing of the same partition").
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/assignment/instance/InstanceReplicaGroupPartitionSelector.java:
##########
@@ -421,8 +459,9 @@ private void replicaGroupBasedMinimumMovement(Map<Integer, List<InstanceConfig>>
List<List<String>> partitionIdToInstancesMap = new ArrayList<>(numPartitions);
List<Set<String>> partitionIdToInstanceSetMap = new ArrayList<>(numPartitions);
- List<List<String>> partitionIdToExistingInstancesMap = new ArrayList<>(existingNumPartitions);
- for (int partitionId = 0; partitionId < numPartitions; partitionId++) {
+ List<List<String>> partitionIdToExistingInstancesMap = new ArrayList<>(partitionIds.size());
+ for (int idx = 0; idx < partitionIds.size(); idx++) {
+ int partitionId = partitionIds.get(idx);
Review Comment:
The code uses an index-based loop to iterate over `partitionIds` and maintain parallel lists (`partitionIdToInstancesMap`, `partitionIdToInstanceSetMap`, `partitionIdToExistingInstancesMap`). This pattern is error-prone when the lists are accessed later. Consider using a Map<Integer, PartitionData> structure where PartitionData holds instances, instanceSet, and existingInstances keyed by partitionId for clearer intent and safer access.
```suggestion
for (Integer partitionId : partitionIds) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Add Kafka ingestion support for subset partitions [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17587:
URL: https://github.com/apache/pinot/pull/17587#issuecomment-3806501636
### :x: 2 Tests Failed:
| Tests completed | Failed | Passed | Skipped |
|---|---|---|---|
| 8818 | 2 | 8816 | 13 |
<details><summary>View the full list of 2 :snowflake: flaky test(s)</summary>
> <pre><code class="language-python">org.apache.pinot.core.operator.transform.function.EqualsTransformFunctionTest::testBinaryOperatorTransformFunction</code></pre>
> **Flake rate in main:** 89.74% (Passed 8 times, Failed 70 times)
> <details><summary>Stack Traces | 0.847s run time</summary>
>
> > <pre><code class="language-python">expected [true] but found [false]</code></pre>
>
> </details>
> <pre><code class="language-python">org.apache.pinot.core.operator.transform.function.EqualsTransformFunctionTest::testBinaryOperatorTransformFunctionNoDict</code></pre>
> **Flake rate in main:** 89.74% (Passed 8 times, Failed 70 times)
> <details><summary>Stack Traces | 0.028s run time</summary>
>
> > <pre><code class="language-python">expected [true] but found [false]</code></pre>
>
> </details>
</details>
To view more test analytics, go to the [Test Analytics Dashboard](https://app.codecov.io/gh/apache/pinot/tests/xiangfu0%3Akafka-subset-partitions)
<sub>📋 Got 3 mins? [Take this short survey](https://forms.gle/22i53Qa1CySZjA6c7) to help us improve Test Analytics.</sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org