FrankChen021 commented on code in PR #20474:
URL: https://github.com/apache/druid/pull/20474#discussion_r4226423380
##########
web-console/src/druid-models/ingestion-spec/ingestion-spec.tsx:
##########
@@ -1335,6 +1347,16 @@ export function issueWithIoConfig(
case 'kafka':
if (!ioConfig.topic && !ioConfig.topicPattern) return 'must have a topic
or topicPattern';
+ if (ioConfig.partitionIds != null) {
+ if (ioConfig.topicPattern) return 'partitionIds requires a single
topic';
+ if (
+ !Array.isArray(ioConfig.partitionIds) ||
+ !ioConfig.partitionIds.length ||
+ ioConfig.partitionIds.some(id => !Number.isInteger(id) || id < 0)
Review Comment:
Fixed in d6c219a7c4. The console now accepts numbers in 0..2147483647 and
decimal integer strings in the same range, and `IoConfig.partitionIds` is typed
`(number | string)[]`. Tests cover `"1"`, `2147483647`, `2147483648`,
`"2147483648"`, `"abc"`, `"-1"` and `"1.5"`.
##########
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:
Thanks, reproduced with a test
(`testBoundedSupervisorReportsLagOnlyForBoundedPartitions`): a bounded config
covering partition 0 of a 3-partition topic made
`updatePartitionLagFromStream()` throw `NoOffsetForPartition` for partitions 1
and 2. Fixed in f9143b986b. Both Kafka lag methods now take their partitions
from `getPartitionsForLag()`, which returns the partitions the supervisor
tracks (from the bounded config) when it is bounded, and the discovered topic
partitions otherwise. I did not let the backfill carry the selection, so the
`boundedStreamConfig` exclusion is unchanged.
##########
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:
Fixed in d73570f29a. `defined` is now `type === 'kafka' && (Boolean(topic)
|| partitionIds != null)`, so the field stays visible while it holds a value
and can be cleared after switching to `topicPattern`. There is a test for it in
`ingestion-spec.spec.ts`.
##########
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:
Agreed that `map(Number)` was too loose. Fixed in d73570f29a, but I kept the
adjustment instead of removing it, so the spec JSON stays numeric (`[0, 2]`
rather than `["0","2"]`). Only plain decimal digit strings are converted;
anything else (`1e3`, `0x10`, `-1`, `abc`) stays a string and is rejected by
`issueWithIoConfig`, as on the server. Tests cover this.
##########
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:
Agreed. Removed in d73570f29a: the extra `KafkaSupervisorIOConfig`
constructor and the two `null`-passing `KafkaRecordSupplier` overloads. The
call sites (`KafkaIndexTask` and the tests) now pass `partitionIds` explicitly.
##########
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:
Agreed, and I checked the other readers of that getter (task creation,
bounded completion, highest-offsets report, idle check, backfill start
offsets): all only need the selected partitions, and resets do not go through
it. Done in 8f3b6b3cd0. `KafkaSupervisor.getOffsetsFromMetadataStorage()` now
also drops excluded partitions, and
`getOffsetsFromMetadataStorageForCurrentPartitions`, its base-class hook, and
my changes to `SupervisorManager` are gone. The filter in
`getHighestCurrentOffsets` stays because publishing tasks can still carry
excluded partitions.
--
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]
