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/06/05 03:09:18 UTC

[PR] Fix Kafka metadata for sparse partition recovery [pinot]

xiangfu0 opened a new pull request, #18687:
URL: https://github.com/apache/pinot/pull/18687

   ## Summary
   - Fix Kafka stream metadata computation to enumerate actual Kafka partition ids instead of using the current Pinot status list size as the next partition id.
   - Preserve existing partition offsets when Pinot has current status, and fetch a stream offset for Kafka partitions that exist in Kafka but are missing from Pinot LLC metadata.
   - Add regression coverage for sparse current statuses in both Kafka 3.0 and Kafka 4.0 providers.
   
   ## User Manual
   No table config change is required for Kafka realtime tables.
   
   After upgrading the controller and Kafka stream plugin code, wait for the scheduled `RealtimeSegmentValidationManager` run or trigger realtime validation through the existing operational path. If Kafka still has the missing partitions, Pinot can recreate missing consuming segments for partitions that exist in Kafka but no longer have consuming/online segments or latest LLC ZK metadata.
   
   Recovered partitions start from the offset selected by validation for new partition repair, typically the smallest currently available Kafka offset. Data older than Kafka retention cannot be recovered by this repair.
   
   ## Sample Table Config
   Existing Kafka realtime configs continue to work. No new config key is required.
   
   ```json
   {
     "tableName": "asset",
     "tableType": "REALTIME",
     "ingestionConfig": {
       "streamIngestionConfig": {
         "streamConfigMaps": [
           {
             "streamType": "kafka",
             "stream.kafka.topic.name": "asset",
             "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
             "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
             "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder"
           }
         ]
       }
     }
   }
   ```
   
   ## Why This Fixes The Issue
   Before this change, the Kafka no-partition-subset path delegated to the SPI default implementation. That implementation first copied current Pinot statuses and then added "new" partition ids using:
   
   ```java
   for (int i = partitionGroupConsumptionStatuses.size(); i < partitionCount; i++)
   ```
   
   For sparse statuses, for example Kafka partitions `0..7` with Pinot statuses for `0,1,3,4,5,6,7`, this produced `[0,1,3,4,5,6,7,7]`: partition `2` stayed missing and partition `7` was duplicated in metadata.
   
   The Kafka providers now fetch actual Kafka partition ids and key current statuses by stream partition id, so the metadata list becomes `[0,1,2,3,4,5,6,7]` and realtime validation can call `setupNewPartitionGroup()` for the missing id.
   
   ## Test Plan
   - `./mvnw spotless:apply checkstyle:check license:format license:check -pl pinot-controller,pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0,pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0`
   - `./mvnw checkstyle:check -pl pinot-controller`
   - `./mvnw -pl pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0,pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0 -Dtest=KafkaStreamMetadataProviderTest#testComputePartitionGroupMetadataRecoversMissingLowPartitionId -Dsurefire.failIfNoSpecifiedTests=false test`
   - `./mvnw -pl pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0 -Dtest=KafkaPartitionLevelConsumerTest#testComputePartitionGroupMetadataUsesKafkaPartitionIds -Dsurefire.failIfNoSpecifiedTests=false test`
   
   Known local limitation: running the Kafka 4.0 embedded `KafkaPartitionLevelConsumerTest` in this workspace is blocked by missing Docker/Testcontainers support (`/var/run/docker.sock` not found). The non-Docker Kafka 4.0 provider regression passes.
   


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4628016499

## Review notes

Traced the change end-to-end (Kafka providers → `PinotLLCRealtimeSegmentManager` consumers) and cross-checked with an independent reviewer. **No CRITICAL/MAJOR issues** — the fix is correct, scoped to the real root cause, preserves steady-state behavior, and has a genuine regression test. Three MINOR follow-ups noted below.

### Why the fix is correct

The empty-subset branch used to delegate to the SPI default, which enumerates new partitions with `for (i = statuses.size(); i < partitionCount; i++)` — it treats the **status-list count** as the next partition id. With a missing low id (a partition that lost all its segments), that yields e.g. `[0,1,3,4,5,6,7,7]`: partition `2` omitted, `7` duplicated.

The new code keys current statuses by `getStreamPartitionGroupId()`, enumerates the **actual** Kafka ids from `fetchPartitionIds()` (sorted), and per id reuses the status `endOffset` if present else fetches `fetchStreamPartitionOffset(offsetCriteria)` → `[0..7]`. The recovered id then flows into `setupNewPartitionGroup` (`PinotLLCRealtimeSegmentManager.java`, "Set up new partitions if not exist" loop), so the consuming segment is actually recreated.

One precision note on impact: the manifestation is **silent non-recovery, not an NPE**. The offset-repair path is guarded by `containsKey` before `selectStartOffset`, and the smallest-offset map is built with `forceGetOffsetFromStream=true` (empty statuses → contiguous `[0..N-1]`, never sparse-broken). So the bug omits the hole id from the new-partition setup loop, which then never runs for it.

### Parity / safety verified

- **Steady state unchanged**: when every partition has a status, the new path emits the same `(id, endOffset)` pairs, just sorted by id rather than status-list order. All controller consumers key by partition id, so ordering is irrelevant; no provider connections are opened when there are no holes.
- **Multi-topic**: keys on `getStreamPartitionGroupId()` (raw stream id), matching `fetchPartitionIds()` space; `PartitionGroupMetadataFetcher` re-encodes raw→Pinot ids afterward — same contract the old SPI default relied on.
- **`forceGetOffsetFromStream`**: Kafka doesn't override the 5-arg overload, so the repair path still passes an empty list → all partitions fetched from stream. Controller change is comment-only.
- **Resources / imports / concurrency**: per-partition providers are try-with-resources closed; required `java.util` + `StreamConsumerFactory*` imports present in both 3.0 and 4.0; `_partitionIdSubset` is read-only; no new shared mutable state.

### Minor follow-ups (non-blocking)

1. **Test hygiene** — `testComputePartitionGroupMetadataUsesKafkaPartitionIds` (both modules) constructs `new KafkaStreamMetadataProvider(...)` without try-with-resources, leaking the consumer. The new `KafkaStreamMetadataProviderTest` does this correctly; worth matching.
2. **Per-partition fetch fan-out** — in the force-from-stream / empty-status path, each hole/new partition opens+closes a metadata provider sequentially. Identical to the prior SPI-default cost (not a regression), but since the provider already holds a connected `_consumer`, a single `beginningOffsets(allTopicPartitions)` could collapse it to one RPC. Reasonable later optimization.
3. **Vanished-partition parity** — the new path emits only ids returned by `fetchPartitionIds()`, so a partition present in the status list but absent from the live stream is dropped (the old code re-emitted it). Only reachable if a topic is deleted+recreated smaller; treating a non-existent partition as EOL is arguably more correct. Awareness only.

### Tests

`KafkaStreamMetadataProviderTest` is a true regression test — it asserts the old size-based algorithm yields `[0,1,3,4,5,6,7,7]`, then asserts the fix yields `[0..7]` **with** the distinguishing offsets (fresh smallest `1002` for the recovered hole vs reused `endOffset` `13` for an existing partition). Added to both 3.0 and 4.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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.

Jackie-Jiang commented on code in PR #18687:
URL: https://github.com/apache/pinot/pull/18687#discussion_r3375905862


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -124,8 +124,31 @@ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientI
       List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
       throws IOException, java.util.concurrent.TimeoutException {
     if (_partitionIdSubset.isEmpty()) {
-      return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,

Review Comment:
   We should fix this method instead. Ideally, we should also pass `_partitionIdSubset` into the method. No reason to duplicate the code



-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4635605066

Yes, this is based on the real issue we saw, but the missing partition is not missing from Kafka.

Kafka still has the partitions. The sparse state is in Pinot LLC metadata: for some existing Kafka partition ids, Pinot no longer has consuming/online segments or latest LLC metadata in ZK. In that state, RVM should be able to recreate consuming segments because the existing validation flow sets up partition groups that exist in stream metadata but are missing in Pinot metadata.

The bug is that Kafka's no-subset metadata path used the SPI default, which derives new ids from `currentStatuses.size()`. That works for the normal Kafka partition expansion case, e.g. `8 -> 16`, but fails when Pinot's current status list is sparse for existing Kafka ids. Example: Kafka has `0..7`, Pinot metadata has `0,1,3,4,5,6,7`. The old computation returns `0,1,3,4,5,6,7,7`, so partition `2` never reaches RVM's `setupNewPartitionGroup` path.

This PR does not assume Kafka partitions disappear intermittently. It asks Kafka for the actual live partition ids and gives RVM correct metadata for partitions that still exist in Kafka but are missing from Pinot metadata.

--
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.

Copilot commented on code in PR #18687:
URL: https://github.com/apache/pinot/pull/18687#discussion_r3360192257


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java:
##########
@@ -282,6 +283,39 @@ public void testSubsetPartitionConsumption()
     }
   }
 
+  @Test
+  public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+      throws Exception {
+    String streamType = "kafka";
+    String streamKafkaBrokerList = _kafkaBrokerAddress;
+    String clientId = "partition-id-client";
+    String tableNameWithType = "tableName_REALTIME";
+
+    Map<String, String> streamConfigMap = new HashMap<>();
+    streamConfigMap.put("streamType", streamType);
+    streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_SUBSET_PARTITION);
+    streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+    StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+    List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+        new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"));
+
+    KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
+    List<PartitionGroupMetadata> partitionGroupMetadataList =
+        streamMetadataProvider.computePartitionGroupMetadata(clientId, streamConfig, currentStatuses, 10000);
+
+    assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+        .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+  }

Review Comment:
   `KafkaStreamMetadataProvider` is `Closeable` (implements `StreamMetadataProvider`), but this new test doesn't close it. Please wrap it in try-with-resources to avoid leaking Kafka consumer/admin-client resources between tests.



##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java:
##########
@@ -286,6 +287,39 @@ public void testSubsetPartitionConsumption() throws TimeoutException {
     }
   }
 
+  @Test
+  public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+      throws Exception {
+    String streamType = "kafka";
+    String streamKafkaBrokerList = _kafkaBrokerAddress;
+    String clientId = "partition-id-client";
+    String tableNameWithType = "tableName_REALTIME";
+
+    Map<String, String> streamConfigMap = new HashMap<>();
+    streamConfigMap.put("streamType", streamType);
+    streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_SUBSET_PARTITION);
+    streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+    StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+    List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+        new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"));
+
+    KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
+    List<PartitionGroupMetadata> partitionGroupMetadataList =
+        streamMetadataProvider.computePartitionGroupMetadata(clientId, streamConfig, currentStatuses, 10000);
+
+    assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+        .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+  }

Review Comment:
   `KafkaStreamMetadataProvider` implements `Closeable` (via `StreamMetadataProvider`), but the new test creates one without closing it. This can leak embedded Kafka resources across tests and cause flakiness; use try-with-resources here (as the new `KafkaStreamMetadataProviderTest` does).



-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4661716184

   Opened the corresponding docs update PR: pinot-contrib/pinot-docs#864


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4654875277

   Updated the implementation to keep the superclass logic count-based. The default `StreamMetadataProvider.computePartitionGroupMetadata()` now fetches upstream `partitionCount`, loops partition ids `0..partitionCount-1`, and looks up existing statuses by `getStreamPartitionGroupId()`. That keeps the logic simple and matches Kafka contiguous partition ids while fixing the sparse-status bug caused by starting at `partitionGroupConsumptionStatuses.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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.

Copilot commented on code in PR #18687:
URL: https://github.com/apache/pinot/pull/18687#discussion_r3360226985


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumerTest.java:
##########
@@ -286,6 +287,39 @@ public void testSubsetPartitionConsumption() throws TimeoutException {
     }
   }
 
+  @Test
+  public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+      throws Exception {
+    String streamType = "kafka";
+    String streamKafkaBrokerList = _kafkaBrokerAddress;
+    String clientId = "partition-id-client";
+    String tableNameWithType = "tableName_REALTIME";
+
+    Map<String, String> streamConfigMap = new HashMap<>();
+    streamConfigMap.put("streamType", streamType);
+    streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_SUBSET_PARTITION);
+    streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+    StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+    List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+        new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"));
+
+    KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
+    List<PartitionGroupMetadata> partitionGroupMetadataList =
+        streamMetadataProvider.computePartitionGroupMetadata(clientId, streamConfig, currentStatuses, 10000);
+
+    assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+        .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+  }

Review Comment:
   `KafkaStreamMetadataProvider` holds a Kafka consumer (via `KafkaPartitionLevelConnectionHandler.close()`), but this new test instantiates it without closing. That can leak threads/connections and make the test suite flaky; wrap it in try-with-resources.



##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumerTest.java:
##########
@@ -282,6 +283,39 @@ public void testSubsetPartitionConsumption()
     }
   }
 
+  @Test
+  public void testComputePartitionGroupMetadataUsesKafkaPartitionIds()
+      throws Exception {
+    String streamType = "kafka";
+    String streamKafkaBrokerList = _kafkaBrokerAddress;
+    String clientId = "partition-id-client";
+    String tableNameWithType = "tableName_REALTIME";
+
+    Map<String, String> streamConfigMap = new HashMap<>();
+    streamConfigMap.put("streamType", streamType);
+    streamConfigMap.put("stream.kafka.topic.name", TEST_TOPIC_SUBSET_PARTITION);
+    streamConfigMap.put("stream.kafka.broker.list", streamKafkaBrokerList);
+    streamConfigMap.put("stream.kafka.consumer.factory.class.name", getKafkaConsumerFactoryName());
+    streamConfigMap.put("stream.kafka.decoder.class.name", "decoderClass");
+    StreamConfig streamConfig = new StreamConfig(tableNameWithType, streamConfigMap);
+
+    List<PartitionGroupConsumptionStatus> currentStatuses = List.of(
+        new PartitionGroupConsumptionStatus(0, 0, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(1, 1, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(3, 3, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(4, 4, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(5, 5, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(6, 6, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"),
+        new PartitionGroupConsumptionStatus(7, 7, new LongMsgOffset(0), new LongMsgOffset(10), "DONE"));
+
+    KafkaStreamMetadataProvider streamMetadataProvider = new KafkaStreamMetadataProvider(clientId, streamConfig);
+    List<PartitionGroupMetadata> partitionGroupMetadataList =
+        streamMetadataProvider.computePartitionGroupMetadata(clientId, streamConfig, currentStatuses, 10000);
+
+    assertEquals(partitionGroupMetadataList.stream().map(PartitionGroupMetadata::getPartitionGroupId)
+        .collect(Collectors.toList()), List.of(0, 1, 2, 3, 4, 5, 6, 7));
+  }

Review Comment:
   `KafkaStreamMetadataProvider` implements `Closeable` (inherited `close()` shuts down the underlying Kafka consumer). This new test creates an instance and uses it without closing, which can leak resources and lead to flaky tests; use try-with-resources here.



-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4655084831

   Adjusted per latest discussion: the 5-arg `StreamMetadataProvider.computePartitionGroupMetadata(...)` overload is back in the SPI base interface as the shared `forceGetOffsetFromStream` wrapper. Kafka 3.0/4.0, Pulsar, and the fake stream provider no longer duplicate that method; they only implement the provider-specific 4-arg partition mapping. Kinesis keeps its existing 5-arg override because it intentionally ignores the force flag and preserves its current behavior.
   
   The Kafka fix remains in the concrete 4-arg methods: fetch upstream partition count, loop `0..partitionCount-1`, key current statuses by `getStreamPartitionGroupId()`, and fetch stream offsets for missing ids. The focused SPI/Kafka/Pulsar reactor passed after this adjustment.


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on code in PR #18687:
URL: https://github.com/apache/pinot/pull/18687#discussion_r3376832545


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java:
##########
@@ -124,8 +124,31 @@ public List<PartitionGroupMetadata> computePartitionGroupMetadata(String clientI
       List<PartitionGroupConsumptionStatus> partitionGroupConsumptionStatuses, int timeoutMillis)
       throws IOException, java.util.concurrent.TimeoutException {
     if (_partitionIdSubset.isEmpty()) {
-      return StreamMetadataProvider.super.computePartitionGroupMetadata(clientId, streamConfig,

Review Comment:
   Agreed. I moved the no-subset fix into `StreamMetadataProvider.computePartitionGroupMetadata()` and reverted Kafka 3.0/4.0 no-subset handling back to `StreamMetadataProvider.super.computePartitionGroupMetadata(...)`. The default now uses `fetchPartitionIds()` when supported, keyed by `getStreamPartitionGroupId()`, and only falls back to contiguous `fetchPartitionCount()` ids for legacy providers. The Kafka partition-subset branch still stays explicit because it must honor `_partitionIdSubset`. Added an SPI regression for sparse current statuses plus the Kafka provider tests for sparse metadata and topic expansion.



-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 merged PR #18687:
URL: https://github.com/apache/pinot/pull/18687


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4628035039

   Follow-up status:
   
   1. Test hygiene: addressed in `a7ea04e720` by wrapping the new `KafkaStreamMetadataProvider` instances in `KafkaPartitionLevelConsumerTest` with try-with-resources for both Kafka 3.0 and 4.0.
   2. Per-partition fetch fan-out: intentionally left as a later optimization because this PR preserves the previous SPI-default cost model and keeps the bug fix scoped.
   3. Vanished-partition parity: intentionally left unchanged; Kafka partitions normally do not shrink, and treating a partition absent from live Kafka metadata as unavailable/end-of-life is acceptable for this scoped recovery fix.
   
   All review threads are resolved/outdated after the latest push.


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.

codecov-commenter commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4628375438

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18687?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :white_check_mark: All modified and coverable lines are covered by tests.
   :white_check_mark: Project coverage is 56.86%. Comparing base ([`fe761d2`](https://app.codecov.io/gh/apache/pinot/commit/fe761d249a272574e9de77575e44102ef4bb9c1f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a7ea04e`](https://app.codecov.io/gh/apache/pinot/commit/a7ea04e72097a6c71e7293fcbf2f4657de95ee28?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   > :exclamation:  There is a different number of reports uploaded between BASE (fe761d2) and HEAD (a7ea04e). Click for more details.
   > 
   > <details><summary>HEAD has 16 uploads less than BASE</summary>
   >
   >| Flag | BASE (fe761d2) | HEAD (a7ea04e) |
   >|------|------|------|
   >|java-21|5|1|
   >|unittests|2|1|
   >|temurin|5|1|
   >|unittests2|1|0|
   >|integration|3|0|
   >|integration1|1|0|
   >|custom-integration1|1|0|
   >|integration2|1|0|
   ></details>
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #18687      +/-   ##
   ============================================
   - Coverage     64.48%   56.86%   -7.62%     
   + Complexity     1291        1    -1290     
   ============================================
     Files          3371     2583     -788     
     Lines        208552   150103   -58449     
     Branches      32570    24261    -8309     
   ============================================
   - Hits         134483    85361   -49122     
   + Misses        63273    57512    -5761     
   + Partials      10796     7230    -3566     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-21](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-7.62%)` | :arrow_down: |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-7.62%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-7.62%)` | :arrow_down: |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <ø> (-0.04%)` | :arrow_down: |
   | [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18687/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/18687?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4635839880

   Added a focused Kafka provider regression for the normal topic expansion case in `56d16e540c`.
   
   The new test covers Kafka reporting partitions `0..7` while Pinot has contiguous current statuses for `0..3`. It verifies the provider returns `0..7`, reuses existing end offsets for `0..3`, and fetches stream offsets for new partitions `4..7`. Added for both Kafka 3.0 and Kafka 4.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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4654959282

   Updated per discussion: the 4-arg `StreamMetadataProvider.computePartitionGroupMetadata(...)` is now abstract, so each stream provider owns its partition-group mapping. Kafka 3.0/4.0 now implement the contiguous Kafka partition logic directly: fetch partition count, loop `0..partitionCount-1`, key current statuses by `getStreamPartitionGroupId()`, and fetch offsets for missing ids. The 5-arg overload remains only as the existing `forceGetOffsetFromStream` wrapper and dispatches to the concrete 4-arg method.


-- 
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] Fix Kafka metadata for sparse partition recovery [pinot]

Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.

xiangfu0 commented on PR #18687:
URL: https://github.com/apache/pinot/pull/18687#issuecomment-4655006263

   Follow-up update: both `StreamMetadataProvider.computePartitionGroupMetadata(...)` overloads are now abstract. Kafka 3.0/4.0, Pulsar, Kinesis, and the fake stream provider all explicitly implement the 5-arg overload now. For Kafka the 5-arg method preserves the previous `forceGetOffsetFromStream` behavior by calling the Kafka-owned 4-arg implementation with either current statuses or an empty status list.


-- 
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