You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/04/30 19:34:20 UTC
(pinot) branch master updated: Support range syntax in stream.kafka.partition.ids config (#18385)
This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 2366a9e6a8b Support range syntax in stream.kafka.partition.ids config (#18385)
2366a9e6a8b is described below
commit 2366a9e6a8ba95e89be39875140fe507d2344426
Author: Abhishek Bafna <ab...@startree.ai>
AuthorDate: Fri May 1 01:04:13 2026 +0530
Support range syntax in stream.kafka.partition.ids config (#18385)
* Support range syntax in stream.kafka.partition.ids config
Add inclusive range notation (e.g. "0-399") to the partition ID parser,
allowing mixed format like "0-99,200,300-399". This makes it practical to
configure partition subsets for high-partition-count topics in logical table
setups without listing hundreds of individual IDs.
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
* Address PR review: add 10K partition ID cap, fix overflow, improve error messages
- Add MAX_TOTAL_PARTITION_IDS (10,000) guard with early bail-out in the
main parsing loop (covers both range and individual ID paths)
- Pre-check range size using long arithmetic before expansion to prevent
integer overflow when end approaches Integer.MAX_VALUE
- Use long loop counter to avoid infinite loop on MAX_VALUE boundary
- Include offending token in parseSingleId error messages for consistency
with parseRange
- Rename test to better reflect its intent (negative ID not misread as range)
- Add tests: exact boundary acceptance, single range overflow, multiple
range overflow, individual ID overflow
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
---------
Co-authored-by: abhishekbafna <ab...@startree.ai>
Co-authored-by: Claude Opus 4.6 <no...@anthropic.com>
---
.../stream/kafka/KafkaPartitionSubsetUtils.java | 83 +++++++++++++---
.../stream/kafka/KafkaStreamConfigProperties.java | 4 +-
.../kafka/KafkaPartitionSubsetUtilsTest.java | 110 +++++++++++++++++++++
3 files changed, 185 insertions(+), 12 deletions(-)
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java
index e9fc7ad69e1..c97dac19c96 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtils.java
@@ -34,20 +34,32 @@ import org.apache.commons.lang3.StringUtils;
*/
public final class KafkaPartitionSubsetUtils {
+ static final int MAX_TOTAL_PARTITION_IDS = 10_000;
+
private KafkaPartitionSubsetUtils() {
}
/**
- * Reads the optional comma-separated partition ID list from the stream config map.
+ * Reads the optional partition ID specification from the stream config map.
+ * Supports three formats, which can be mixed in a single value:
+ * <ul>
+ * <li>Individual IDs: {@code "0,2,5"}</li>
+ * <li>Inclusive ranges (both start and end are included): {@code "0-399"}</li>
+ * <li>Mixed: {@code "0-99,200,300-399"}</li>
+ * </ul>
* Returns a sorted, deduplicated list for stable ordering when used for partition group metadata.
* Duplicate IDs in the config are silently removed; this ensures stable ordering and prevents
* duplicate processing of the same partition.
*
+ * <p>The total number of unique partition IDs must not exceed {@link #MAX_TOTAL_PARTITION_IDS}
+ * (currently 10,000). This guard prevents accidental OOM from typos or overly broad ranges.
+ *
* @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
+ * @throws IllegalArgumentException if the value contains invalid entries or exceeds
+ * {@link #MAX_TOTAL_PARTITION_IDS}
*/
@Nullable
public static List<Integer> getPartitionIdsFromConfig(Map<String, String> streamConfigMap) {
@@ -63,16 +75,16 @@ public final class KafkaPartitionSubsetUtils {
if (trimmed.isEmpty()) {
continue;
}
- try {
- int partitionId = Integer.parseInt(trimmed);
- if (partitionId < 0) {
- throw new IllegalArgumentException("Invalid " + key
- + " value: partition IDs must be non-negative, got '" + value + "'");
- }
- idSet.add(partitionId);
- } catch (NumberFormatException e) {
+ int hyphenIndex = trimmed.indexOf('-', 1);
+ if (hyphenIndex > 0) {
+ parseRange(trimmed, hyphenIndex, key, value, idSet);
+ } else {
+ parseSingleId(trimmed, key, value, idSet);
+ }
+ if (idSet.size() > MAX_TOTAL_PARTITION_IDS) {
throw new IllegalArgumentException(
- "Invalid " + key + " value: expected comma-separated integers, got '" + value + "'", e);
+ "Invalid " + key + " value: total partition count " + idSet.size()
+ + " exceeds maximum allowed " + MAX_TOTAL_PARTITION_IDS + ", got '" + value + "'");
}
}
if (idSet.isEmpty()) {
@@ -82,4 +94,53 @@ public final class KafkaPartitionSubsetUtils {
Collections.sort(ids);
return ids;
}
+
+ private static void parseSingleId(String trimmed, String key, String originalValue, Set<Integer> idSet) {
+ try {
+ int partitionId = Integer.parseInt(trimmed);
+ if (partitionId < 0) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: partition ID must be non-negative in '" + trimmed + "', got '"
+ + originalValue + "'");
+ }
+ idSet.add(partitionId);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: expected integer or range in '" + trimmed + "', got '" + originalValue + "'",
+ e);
+ }
+ }
+
+ // Parses an inclusive range token like "0-399" (both start and end are included).
+ private static void parseRange(String trimmed, int hyphenIndex, String key, String originalValue,
+ Set<Integer> idSet) {
+ String startStr = trimmed.substring(0, hyphenIndex).trim();
+ String endStr = trimmed.substring(hyphenIndex + 1).trim();
+ int start;
+ int end;
+ try {
+ start = Integer.parseInt(startStr);
+ end = Integer.parseInt(endStr);
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: invalid range '" + trimmed + "', got '" + originalValue + "'", e);
+ }
+ if (start < 0 || end < 0) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: partition IDs must be non-negative, got '" + originalValue + "'");
+ }
+ if (start > end) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: range start must be <= end in '" + trimmed + "', got '" + originalValue + "'");
+ }
+ long rangeSize = (long) end - start + 1;
+ if (idSet.size() + rangeSize > MAX_TOTAL_PARTITION_IDS) {
+ throw new IllegalArgumentException(
+ "Invalid " + key + " value: total partition count exceeds maximum allowed " + MAX_TOTAL_PARTITION_IDS
+ + ", got '" + originalValue + "'");
+ }
+ for (long i = start; i <= end; i++) {
+ idSet.add((int) i);
+ }
+ }
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
index 1d7fa19de84..9614f436fbf 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/main/java/org/apache/pinot/plugin/stream/kafka/KafkaStreamConfigProperties.java
@@ -70,8 +70,10 @@ public class KafkaStreamConfigProperties {
public static final String KAFKA_CONSUMER_PROP_PREFIX = "kafka.consumer.prop";
/**
- * Optional comma-separated list of Kafka partition IDs to consume (e.g. "0,2,5").
+ * Optional comma-separated list of Kafka partition IDs or inclusive ranges to consume
+ * (e.g. "0,2,5" or "0-399" or "0-99,200,300-399").
* When set, only these partitions are used for the table; when absent, all topic partitions are consumed.
+ * The total number of unique partition IDs must not exceed 10,000.
*/
public static final String PARTITION_IDS = "kafka.partition.ids";
}
diff --git a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtilsTest.java b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtilsTest.java
index 9b262fab056..07b7f45b05c 100644
--- a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtilsTest.java
+++ b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-base/src/test/java/org/apache/pinot/plugin/stream/kafka/KafkaPartitionSubsetUtilsTest.java
@@ -198,4 +198,114 @@ public class KafkaPartitionSubsetUtilsTest {
"0,1,-5,2");
KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
}
+
+ @Test
+ public void testRangeMixedWithIndividualIds() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0-4,50,100-104");
+ List<Integer> result = KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(result, List.of(0, 1, 2, 3, 4, 50, 100, 101, 102, 103, 104));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testRangeStartGreaterThanEnd() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "10-5");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testRangeEmptyEnd() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "5-");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testNegativePartitionIdNotMisreadAsRange() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "-1,0,1");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testRangeNonNumeric() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "abc-def");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testRangeNegativeStart() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "-1-5");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test
+ public void testSingleElementRange() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS), "5-5");
+ List<Integer> result = KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(result, List.of(5));
+ }
+
+ @Test
+ public void testOverlappingRangesAreDeduplicated() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0-5,3-8");
+ List<Integer> result = KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(result, List.of(0, 1, 2, 3, 4, 5, 6, 7, 8));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testSingleRangeExceedsTotalMax() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0-" + KafkaPartitionSubsetUtils.MAX_TOTAL_PARTITION_IDS);
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testMultipleRangesExceedTotalMax() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0-5999,6000-11999");
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
+
+ @Test
+ public void testExactlyMaxPartitionIdsAccepted() {
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ "0-" + (KafkaPartitionSubsetUtils.MAX_TOTAL_PARTITION_IDS - 1));
+ List<Integer> result = KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ Assert.assertNotNull(result);
+ Assert.assertEquals(result.size(), KafkaPartitionSubsetUtils.MAX_TOTAL_PARTITION_IDS);
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void testIndividualIdsExceedTotalMax() {
+ StringBuilder sb = new StringBuilder();
+ for (int i = 0; i <= KafkaPartitionSubsetUtils.MAX_TOTAL_PARTITION_IDS; i++) {
+ if (i > 0) {
+ sb.append(',');
+ }
+ sb.append(i);
+ }
+ Map<String, String> config = new HashMap<>();
+ config.put(KafkaStreamConfigProperties.constructStreamProperty(KafkaStreamConfigProperties.PARTITION_IDS),
+ sb.toString());
+ KafkaPartitionSubsetUtils.getPartitionIdsFromConfig(config);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org