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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
