amaechler commented on code in PR #20474:
URL: https://github.com/apache/druid/pull/20474#discussion_r4222430059
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java:
##########
@@ -55,9 +60,71 @@ public class KafkaSupervisorIOConfig extends
SeekableStreamSupervisorIOConfig
private final KafkaConfigOverrides configOverrides;
private final String topic;
private final String topicPattern;
+ @Nullable
+ private final ImmutableSortedSet<Integer> partitionIds;
private final boolean emitTimeLagMetrics;
private final KafkaHeaderBasedFilterConfig headerBasedFilterConfig;
+ /**
+ * Retains the signature used before {@code partitionIds} was introduced
(defaults it to null), so that callers
+ * compiled against it keep working.
+ */
+ public KafkaSupervisorIOConfig(
Review Comment:
Nit: this constructor has no callers outside tests, and the same goes for
the two new `null`-passing overloads in `KafkaRecordSupplier`. Adding the
parameter at the existing call sites would drop roughly 70 lines. It would also
stop a future caller from silently losing `partitionIds`.
##########
web-console/src/druid-models/ingestion-spec/ingestion-spec.tsx:
##########
@@ -1156,6 +1157,17 @@ export function
getIoConfigFormFields(ingestionComboType: IngestionComboType): F
placeholder: 'your_kafka_topic',
info: 'The name of the Kafka topic to ingest from.',
},
+ {
+ name: 'partitionIds',
+ label: 'Partition IDs',
+ type: 'string-array',
+ defined: ioConfig => ioConfig.type === 'kafka' &&
Boolean(ioConfig.topic),
+ placeholder: 'Optional; comma-separated, e.g. 0, 2',
+ hideInMore: ioConfig => ioConfig.partitionIds == null,
+ valueAdjustment: partitionIds =>
Review Comment:
Could this `valueAdjustment` just go? The server and
`isValidKafkaPartitionId` already accept integer strings. `map(Number)` also
loosens what the form accepts: `1e3` becomes 1000 and `0x10` becomes 16, while
the server rejects both as strings.
##########
web-console/src/druid-models/ingestion-spec/ingestion-spec.tsx:
##########
@@ -1156,6 +1157,17 @@ export function
getIoConfigFormFields(ingestionComboType: IngestionComboType): F
placeholder: 'your_kafka_topic',
info: 'The name of the Kafka topic to ingest from.',
},
+ {
+ name: 'partitionIds',
+ label: 'Partition IDs',
+ type: 'string-array',
+ defined: ioConfig => ioConfig.type === 'kafka' &&
Boolean(ioConfig.topic),
Review Comment:
If a user fills Partition IDs, then clears `topic` and enters a
`topicPattern`, the Partition IDs field disappears. `issueWithIoConfig` still
returns "partitionIds requires a single topic", so the form is stuck unless
they edit the JSON. Could the field stay visible while it holds a value, or
could `partitionIds` be cleared when `topic` is unset?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -270,6 +335,32 @@ protected
List<SeekableStreamIndexTask<KafkaTopicPartition, Long, KafkaRecordEnt
return taskList;
}
+ /** Keeps offsets for the configured partitions only. */
+ @Override
+ protected Map<KafkaTopicPartition, Long> getHighestCurrentOffsets()
Review Comment:
Optional simplification: `KafkaSupervisor.getOffsetsFromMetadataStorage()`
already filters stored offsets by topic. Adding the selection filter there
would let you drop the new `getOffsetsFromMetadataStorageForCurrentPartitions`
hook in `SeekableStreamSupervisor`, its two call-site changes and this
override. As far as I can tell, every caller only needs the selected
partitions. Offsets for excluded partitions are still kept, because commits and
resets go through the metadata coordinator rather than this getter.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java:
##########
@@ -206,7 +206,8 @@ public KafkaSupervisorSpec createBackfillSpec(
ioConfig.getStopTaskCount(),
ioConfig.isEmitTimeLagMetrics(),
ioConfig.getServerPriorityToReplicas(),
- boundedStreamConfig
+ boundedStreamConfig,
+ null
Review Comment:
Backfilling a supervisor that has `partitionIds` set seems to break the
backfill's lag refresh. `createBackfillSpec` passes `null` for `partitionIds`,
so the bounded supervisor's `updatePartitionLagFromStream` asks
`getPartitionIds` for the whole topic. It then seeks and reads positions for
partitions that bounded mode never assigned.
I reproduced this with a `KafkaSupervisorTest` case: a bounded config
covering partition 0 of a 3-partition topic, then
`updatePartitionLagFromStream()` throws `NoOffsetForPartition` for partitions 1
and 2. In a real cluster `updateCurrentAndLatestOffsets` catches it, so I
suspect the backfill supervisor logs a warning every cycle and never reports
lag (not confirmed end to end).
Options: let the backfill carry the selection, which means relaxing the
`boundedStreamConfig` exclusion for derived specs. Or have the Kafka lag path
use the partitions the supervisor actually tracks instead of re-querying the
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: [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]
