github-advanced-security[bot] commented on code in PR #18466:
URL: https://github.com/apache/druid/pull/18466#discussion_r2324732659
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10277)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10276)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10278)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
+ spec.getIoConfig().getPollTimeout(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10279)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
+ spec.getIoConfig().getPollTimeout(),
+ true,
+ existingTaskGroup.getMinimumMessageTime(),
+ existingTaskGroup.getMaximumMessageTime(),
+ spec.getIoConfig().getInputFormat(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10280)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
+ spec.getIoConfig().getPollTimeout(),
+ true,
+ existingTaskGroup.getMinimumMessageTime(),
+ existingTaskGroup.getMaximumMessageTime(),
+ spec.getIoConfig().getInputFormat(),
+ spec.getIoConfig().getConfigOverrides(),
+ spec.getIoConfig().isMultiTopic(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10282)
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -584,6 +608,218 @@
}
}
+ private Map<PartitionIdType, SequenceOffsetType>
getLatestOffsetsFromMetadataStore()
+ {
+ try {
+ DataSourceMetadata metadata =
indexerMetadataStorageCoordinator.retrieveDataSourceMetadata(dataSource);
+ if (metadata instanceof SeekableStreamDataSourceMetadata) {
+ @SuppressWarnings("unchecked")
+ SeekableStreamDataSourceMetadata<PartitionIdType, SequenceOffsetType>
streamMetadata =
+ (SeekableStreamDataSourceMetadata<PartitionIdType,
SequenceOffsetType>) metadata;
+ return
streamMetadata.getSeekableStreamSequenceNumbers().getPartitionSequenceNumberMap();
+ }
+ }
+ catch (Exception e) {
+ log.warn(e, "Failed to retrieve latest offsets from metadata store,
using current partition state");
+ }
+ return Collections.emptyMap();
+ }
+
+ /**
+ * Handles task count changes for perpetual tasks using updateConfig instead
of graceful shutdown.
+ * This approach pauses tasks, recalculates partition assignments, and sends
config updates.
+ */
+ private boolean changeTaskCountForPerpetualTasks(int desiredActiveTaskCount,
Stopwatch scaleActionStopwatch)
Review Comment:
## Useless parameter
The parameter 'scaleActionStopwatch' is never used.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10275)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
+ spec.getIoConfig().getPollTimeout(),
+ true,
+ existingTaskGroup.getMinimumMessageTime(),
+ existingTaskGroup.getMaximumMessageTime(),
+ spec.getIoConfig().getInputFormat(),
+ spec.getIoConfig().getConfigOverrides(),
+ spec.getIoConfig().isMultiTopic(),
+ spec.getIoConfig().getTaskDuration().getStandardMinutes()
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10283)
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -226,6 +250,73 @@
);
}
+ @Override
+ protected SeekableStreamIndexTaskIOConfig<KafkaTopicPartition, Long>
createUpdatedTaskIoConfig(
+ Set<KafkaTopicPartition> partitions,
+ TaskGroup existingTaskGroup,
+ Map<KafkaTopicPartition, Long> latestCommittedOffsets
+ )
+ {
+ log.info("Creating updated task IO config for task group [%s]",
existingTaskGroup.getId());
+ Map<KafkaTopicPartition, Long> startingSequences = new HashMap<>();
+ Set<KafkaTopicPartition> exclusiveStartSequenceNumberPartitions = new
HashSet<>();
+
+ for (KafkaTopicPartition partition : partitions) {
+ Long offset;
+ if (!latestCommittedOffsets.containsKey(partition)) {
+ log.warn("No committed offset found for partition [%s], using
NOT_SET", partition);
+ offset = NOT_SET;
+ } else {
+ offset = latestCommittedOffsets.get(partition);
+ }
+
+ startingSequences.put(partition, offset);
+ }
+
+ SeekableStreamStartSequenceNumbers<KafkaTopicPartition, Long>
startSequenceNumbers =
+ new SeekableStreamStartSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ startingSequences,
+ exclusiveStartSequenceNumberPartitions
+ );
+
+ // For end sequences, use NOT_SET to indicate open-ended reading
+ Map<KafkaTopicPartition, Long> endingSequences = new HashMap<>();
+ for (KafkaTopicPartition partition : partitions) {
+ endingSequences.put(partition, NOT_SET);
+ }
+
+ SeekableStreamEndSequenceNumbers<KafkaTopicPartition, Long>
endSequenceNumbers =
+ new SeekableStreamEndSequenceNumbers<>(
+ spec.getIoConfig().getStream(),
+ endingSequences
+ );
+
+ log.info(
+ "Created updated IOConfig with starting sequences [%s] for partitions
[%s]",
+ startingSequences, partitions
+ );
+
+ // Create the updated IOConfig
+ return new KafkaIndexTaskIOConfig(
+ existingTaskGroup.getId(),
+ existingTaskGroup.getBaseSequenceName(),
+ null,
+ null,
+ startSequenceNumbers,
+ endSequenceNumbers,
+ spec.getIoConfig().getConsumerProperties(),
+ spec.getIoConfig().getPollTimeout(),
+ true,
+ existingTaskGroup.getMinimumMessageTime(),
+ existingTaskGroup.getMaximumMessageTime(),
+ spec.getIoConfig().getInputFormat(),
+ spec.getIoConfig().getConfigOverrides(),
Review Comment:
## Deprecated method or constructor invocation
Invoking [KafkaSupervisorSpec.getIoConfig](1) should be avoided because it
has been deprecated.
[Show more
details](https://github.com/apache/druid/security/code-scanning/10281)
--
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]
