You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org> on 2025/10/08 23:15:28 UTC
[PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Nikita-Shupletsov opened a new pull request, #20665:
URL: https://github.com/apache/kafka/pull/20665
MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available.
Before we added caching for consumer next offsets we'd called `mainConsumer.position` and always expected something back. When we added the caching, we kept the check that we always have nextOffset, but as the logic changed to fetching the offsets from poll, we may not have anything for topics that have no messages.
This PR accounts for that.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19775: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2426272664
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -470,17 +470,24 @@ private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition)
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ if (offsetAndMetadata == null) {
+ try {
+ offset = mainConsumer.position(partition);
+ leaderEpoch = Optional.empty();
+ } catch (final TimeoutException error) {
+ // the `consumer.position()` call should never block, because we know that we did process data
+ // for the requested partition and thus the consumer should have a valid local position
+ // that it can return immediately
Review Comment:
Is that still valid in the corner case? I think in the case you are describing, we haven't processed any records so we may not have a `position` for the partition yet and this will actually block to do an offset fetch to find the last committed offset for the partition?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415241756
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
+ if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ offset = offsetAndMetadata.offset();
+ leaderEpoch = offsetAndMetadata.leaderEpoch();
+ } else {
+ // it may be that we have not yet consumed any record from this partition, hence nothing to commit
+ return Optional.empty();
}
}
- return new OffsetAndMetadata(offset,
+ return Optional.of(new OffsetAndMetadata(offset,
Review Comment:
we first check partitionGroup.headRecordOffset. if there is no offset there we check nextOffsetsAndMetadataToBeConsumed. `if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {` is inside of the `offset == null` check, so we will need to duplicate that code here and there
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415240316
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
+ if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
Review Comment:
nit: I find the `constainsKey()` followed by `get()` pattern always confusing and hard to read... How about
```
final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
if (offsetAndMetadata != null) {
offset = offsetAndMetadata.offset();
leaderEpoch = offsetAndMetadata.leaderEpoch();
}
```
Just a personal preference.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415241275
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java:
##########
@@ -254,6 +256,64 @@ public void testRegexRecordsAreProcessedAfterNewTopicCreatedWithMultipleSubtopol
}
}
+ @Test
+ public void test() throws Exception {
Review Comment:
```suggestion
public void shouldNotCrashIfPatternMatchesTopicHasNoData() throws Exception {
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415218811
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
+ if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
Review Comment:
the change that reworked that code to use cached offsets: https://github.com/apache/kafka/pull/17091/files#diff-a76674468cda8772230fb8411717cf9068b1a363a792f32c602fb2ec5ba9efd7R472
@aliehsaeedii @lucasbru, as you folks worked on it, could you please take a look? thanks
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415243760
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
+ if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ offset = offsetAndMetadata.offset();
+ leaderEpoch = offsetAndMetadata.leaderEpoch();
+ } else {
+ // it may be that we have not yet consumed any record from this partition, hence nothing to commit
+ return Optional.empty();
}
}
- return new OffsetAndMetadata(offset,
+ return Optional.of(new OffsetAndMetadata(offset,
Review Comment:
Ah. My bad. Makes sense.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415233367
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -505,12 +497,12 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
// input partitions
final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
inputPartitions() : consumedOffsets.keySet();
- committableOffsets = new HashMap<>(partitionsNeedCommit.size());
- for (final TopicPartition partition : partitionsNeedCommit) {
- committableOffsets.put(partition, findOffsetAndMetadata(partition));
- }
- break;
+ return partitionsNeedCommit.stream()
+ .flatMap(partition -> findOffsetAndMetadata(partition)
+ .map(offsetAndMetadata -> Map.entry(partition, offsetAndMetadata))
Review Comment:
If `findOffsetAndMetadata` returns an empty `Optional`, the partition will be dropped, and we does not end up in the computed "committableOffsets" `Map`, right?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19775: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #20665:
URL: https://github.com/apache/kafka/pull/20665
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415243215
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
Review Comment:
Nice slide cleanup.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415235713
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -505,12 +497,12 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
// input partitions
final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
inputPartitions() : consumedOffsets.keySet();
- committableOffsets = new HashMap<>(partitionsNeedCommit.size());
- for (final TopicPartition partition : partitionsNeedCommit) {
- committableOffsets.put(partition, findOffsetAndMetadata(partition));
- }
- break;
+ return partitionsNeedCommit.stream()
+ .flatMap(partition -> findOffsetAndMetadata(partition)
+ .map(offsetAndMetadata -> Map.entry(partition, offsetAndMetadata))
Review Comment:
yes. if `findOffsetAndMetadata` returns an empty optional, that topic partition will not be in the result 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415238036
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
+ if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ offset = offsetAndMetadata.offset();
+ leaderEpoch = offsetAndMetadata.leaderEpoch();
+ } else {
+ // it may be that we have not yet consumed any record from this partition, hence nothing to commit
+ return Optional.empty();
}
}
- return new OffsetAndMetadata(offset,
+ return Optional.of(new OffsetAndMetadata(offset,
Review Comment:
nit: should we move this into the ` if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {` block?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415233643
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -465,38 +465,30 @@ public Map<TopicPartition, OffsetAndMetadata> prepareCommit(final boolean clean)
}
}
- private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition) {
+ private Optional<OffsetAndMetadata> findOffsetAndMetadata(final TopicPartition partition) {
Long offset = partitionGroup.headRecordOffset(partition);
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
- }
- } catch (final KafkaException fatal) {
Review Comment:
as there are no client calls anymore, the catch is redundant
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19775: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20665:
URL: https://github.com/apache/kafka/pull/20665#issuecomment-3409153082
Thanks for the fix! -- Merged to `trunk` and cherry-picked to `4.1` and `4.0` branches.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19775: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2426280921
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -470,17 +470,24 @@ private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition)
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ if (offsetAndMetadata == null) {
+ try {
+ offset = mainConsumer.position(partition);
Review Comment:
Yes, this is also what I understood when digging into it last week. I'm actually not sure that logic is fully watertight - what if my regular expression matches a disjoint subset of topics - none of them will have the metadata, so I have lost it, right? It seems a bit of a best-effort thing to me... In which we could also consider not committing the "partition we never consumed from" 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415235787
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -505,12 +497,12 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
// input partitions
final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
inputPartitions() : consumedOffsets.keySet();
- committableOffsets = new HashMap<>(partitionsNeedCommit.size());
- for (final TopicPartition partition : partitionsNeedCommit) {
- committableOffsets.put(partition, findOffsetAndMetadata(partition));
- }
- break;
+ return partitionsNeedCommit.stream()
+ .flatMap(partition -> findOffsetAndMetadata(partition)
Review Comment:
Java streams API always confused me. Why do we use `flatMap` here? Don't we get a single entry in "committableOffsets" Map that we compute here? Or is it needed, because we might drop some 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415240265
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -505,12 +497,12 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
// input partitions
final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
inputPartitions() : consumedOffsets.keySet();
- committableOffsets = new HashMap<>(partitionsNeedCommit.size());
- for (final TopicPartition partition : partitionsNeedCommit) {
- committableOffsets.put(partition, findOffsetAndMetadata(partition));
- }
- break;
+ return partitionsNeedCommit.stream()
+ .flatMap(partition -> findOffsetAndMetadata(partition)
Review Comment:
so the path here is:
a stream of partitionsNeedCommit -> flatMap(partition -> to offsetAndMetadata optional -> map optional to map.entry optional -> map it to a stream) -> collect
the reason why we map the the optional to a stream is that it looks nicer. otherwise we would need to have two operations: filter(Optonal::ifPresent).map(Optional::get).
if you prefer that syntax, I will update 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2415245426
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -505,12 +497,12 @@ private Map<TopicPartition, OffsetAndMetadata> committableOffsetsAndMetadata() {
// input partitions
final Set<TopicPartition> partitionsNeedCommit = processorContext.processorMetadata().needsCommit() ?
inputPartitions() : consumedOffsets.keySet();
- committableOffsets = new HashMap<>(partitionsNeedCommit.size());
- for (final TopicPartition partition : partitionsNeedCommit) {
- committableOffsets.put(partition, findOffsetAndMetadata(partition));
- }
- break;
+ return partitionsNeedCommit.stream()
+ .flatMap(partition -> findOffsetAndMetadata(partition)
Review Comment:
Subjective. :) If would it personally find easy to read using `filter(...).map(...)`; it's more explicit, and less "magic"
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19775: Don't fail if nextOffsetsAndMetadataToBeConsumed is not available. [kafka]
Posted by "Nikita-Shupletsov (via GitHub)" <gi...@apache.org>.
Nikita-Shupletsov commented on code in PR #20665:
URL: https://github.com/apache/kafka/pull/20665#discussion_r2422215571
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -470,17 +470,24 @@ private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition partition)
Optional<Integer> leaderEpoch = partitionGroup.headRecordLeaderEpoch(partition);
final long partitionTime = partitionGroup.partitionTimestamp(partition);
if (offset == null) {
- try {
- if (nextOffsetsAndMetadataToBeConsumed.containsKey(partition)) {
- final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
- offset = offsetAndMetadata.offset();
- leaderEpoch = offsetAndMetadata.leaderEpoch();
- } else {
- // This indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException("Stream task " + id + " does not know the partition: " + partition);
+ final OffsetAndMetadata offsetAndMetadata = nextOffsetsAndMetadataToBeConsumed.get(partition);
+ if (offsetAndMetadata == null) {
+ try {
+ offset = mainConsumer.position(partition);
Review Comment:
my understanding so far:
we have some metadata that's one per task. but we have multiple input partitions. so instead of picking one, we just add it everywhere. and during restore we just read them all in whatever order and expect them to be more or less the same.
so if we are adding a new partition on the fly, we want that partition to also have that metadata. if we follow that logic.
so if we follow that logic, we need to commit even for that empty partition.
so I added a fallback to the previous logic when we ask the consumer about the offset
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org