KAFKA-8880: Add overloaded function of Consumer.committed #7304

guozhangwang · 2019-09-05T23:48:12Z

Add the overloaded functions.
Update the code in Streams to use the batch API for better latency (this applies to both active StreamsTask for initialize the offsets, as well as the StandbyTasks for updating offset limits).
Also update all unit test to replace the deprecated APIs.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

bbejeck

@guozhangwang just one minor nit, otherwise LGTM

bbejeck · 2019-09-10T16:01:53Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * Get the last committed offsets for the given partitions (whether the commit happened by this process or
+     * another). The returned offsets will be used as the position for the consumer in the event of a failure.
+     * <p>
+     * This call will block to do a remote call to get the latest committed offsets from the server.


nit: Should the description here match https://github.com/apache/kafka/pull/7304/files#diff-267b7c1e68156c1301c56be63ae41dd0R1779-R1782 from above with the exception that the user specifies the timeout in this case.

kamalcph

LGTM, left some nits.

kamalcph · 2019-09-11T11:34:15Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

            if (offsets == null) {
                throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before the last " +
-                        "committed offset for partition " + partition + " could be determined. Try tuning default.api.timeout.ms " +
-                        "larger to relax the threshold.");
+                    "committed offset for partition " + partitions + " could be determined. Try tuning default.api.timeout.ms " +


nit: partition -> partitions

kamalcph · 2019-09-11T11:37:00Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * Get the last committed offsets for the given partitions (whether the commit happened by this process or
+     * another). The returned offsets will be used as the position for the consumer in the event of a failure.
+     * <p>
+     * This call will do a remote call to get the latest committed offset from the server, and will block until the


nit: latest committed offset (or) last committed offset?

Should be latest committed offset

kamalcph · 2019-09-11T11:37:48Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * the caller), or the timeout specified by {@code default.api.timeout.ms} expires (in which case a
+     * {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
+     *
+     * @param partitions The partition to check


nit: partition -> partitions

kamalcph · 2019-09-11T11:39:13Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * {@link org.apache.kafka.common.errors.TimeoutException} is thrown to the caller).
+     *
+     * @param partitions The partition to check
+     * @return The last committed offset and metadata or null if there was no prior commit


It's not clear whether the returned map is null (or) OffsetAndMetadata will be null for the given partition.

Good catch!

kamalcph · 2019-09-11T11:42:22Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * This call will block to do a remote call to get the latest committed offsets from the server.
+     *
+     * @param partitions The partitions to check
+     * @param timeout  The maximum amount of time to await the current committed offset


Not sure. the current -> to get the latest/last

…tch-committed

guozhangwang · 2019-09-15T23:27:37Z

cc @cpettitt-confluent @bbejeck for another look.

cpettitt-confluent

Mostly minor tweaks suggested.

cpettitt-confluent · 2019-09-16T15:26:44Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * If any of the partitions requested do not exist, an exception would be thrown.
+     * <p>
+     * This call will do a remote call to get the latest committed offsets from the server, and will block until the
+     * committed offset is gotten successfully, an unrecoverable error is encountered (in which case it is thrown to


s/offset is/offsets are/

cpettitt-confluent · 2019-09-16T15:28:06Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+    }
+
+    /**
+     * Get the last committed offsets for the given partitions (whether the commit happened by this process or


Minor: Probably obvious, but since this doc is pretty good about being clear on details, maybe it is worth pointing out that this is for the consumer group?

Actually even if the consumer is not part of a group and not using subscribe as well it can still commit offsets, and others can get its committed offsets as long as they know its group.id.

cpettitt-confluent · 2019-09-16T15:31:12Z

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

+     * <p>
+     * Partitions that do not have a committed offset would not be included in the returned map.
+     * <p>
+     * If any of the partitions requested do not exist, an exception would be thrown.


Now that we're batching calls it might be nice to return all of the valid ones we received and some marker for those we did not.

Sans that, we should specify what type of exception you get and it would be nice to be able to get details about which partitions did not exist.

I've thought about that when discussing about KIP-520, e.g. on admin client when getting committed offsets we return a map<topic-partition, future<>> and each future can either be an exception or the actual value. But for consumer we do not have such APIs so I've decided to stick with consistency.

As for the exception, it would indicate which topic-partition(s) do not exist.

Sounds reasonable. Is the non-existant topic partition list accessible programmatically or just in the exception text? The former seems a bit nicer in allowing for potential recovery at runtime.

cpettitt-confluent · 2019-09-16T15:33:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java

@@ -250,16 +251,23 @@ public boolean hasStateStores() {
        return stateMgr.changelogPartitions();
    }

-    long committedOffsetForPartition(final TopicPartition partition) {
+    Map<TopicPartition, Long> committedOffsetForPartition(final Set<TopicPartition> partitions) {


committedOffsetsForPartitions

cpettitt-confluent · 2019-09-16T15:38:26Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

    }

-    long offsetLimit(final TopicPartition partition) {
+    private long offsetLimit(final TopicPartition partition) {


cpettitt-confluent · 2019-09-16T15:41:30Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java

-        final long newLimit = committedOffsetForPartition(partition);
-        final long previousLimit = offsetLimits.put(partition, newLimit);
-        if (previousLimit > newLimit) {
+        if (previousLimit != null && previousLimit > newLimit) {


We should either do this check for all updated limits or none.

Ack. Thinking about this a bit more, I think we do not need the updateableOffsetLimits since it always have the same topic-partitions with offsetLimits, and instead we just keep a flag and always ask for all topic-partitions in offsetLimits. WDYT?

+1 now that we're doing the call in batch.

cpettitt-confluent · 2019-09-16T15:43:00Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java

-                        final long limit) {
-        log.trace("Updating store offset limit for partition {} to {}", partition, limit);
-        offsetLimits.put(partition, limit);
+    void putOffsetLimit(final Map<TopicPartition, Long> offsets) {


putOffsetLimits

…tch-committed

guozhangwang · 2019-09-20T18:45:53Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -481,7 +481,6 @@ void commit(final boolean startNewTransaction, final Map<TopicPartition, Long> p
            final long offset = entry.getValue() + 1;
            final long partitionTime = partitionTimes.get(partition);
            consumedOffsetsAndMetadata.put(partition, new OffsetAndMetadata(offset, encodeTimestamp(partitionTime)));
-            stateMgr.putOffsetLimit(partition, offset);


This line is not needed since for stream task we only need to put offset limit once, which is before the restoration. During normal processing we do not need to set offset limit any more.

+1. That should not be needed.

bbejeck

Still LGTM, just one minor additional comment.

bbejeck · 2019-09-20T20:30:04Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

+                log.debug("A committed timestamp was detected: setting the partition time of partition {}"
+                    + " to {} in stream task {}", partition, committedTimestamp, this);
+            } else {
+                log.debug("No committed timestamp was found in metadata for partition {}", partition);


If there is no metadata would we want to use the latest timestamp seen so far for the StreamTask and use that to set PartitionGroup#setPartitionTime?

I followed the logic from the other PR that @RichardYuSTUG did, would leave to Richard to explain why we did this :)

@bbejeck @guozhangwang Oops, looks like I missed this. Bill has a point here. I will probably log a JIRA to get this done.

guozhangwang · 2019-09-21T01:15:21Z

retest this please

guozhangwang · 2019-09-23T16:26:41Z

retest this please

guozhangwang · 2019-09-24T20:21:57Z

Merging to trunk; will submit a separate PR for docs change.

Conflicts: * .gitignore: addition of clients/src/generated-test was near local additions for support-metrics. * checkstyle/suppressions.xml: upstream refactoring of exclusions for generator were near the local changes for support-metrics. * gradle.properties: scala version bump caused a minor conflict due to the kafka version change locally. gradle/dependencies.gradle: bcpkix version bump was near avro additions in the local version. * apache-github/trunk: (49 commits) KAFKA-8471: Replace control requests/responses with automated protocol (apache#7353) MINOR: Don't generate unnecessary strings for debug logging in FetchSessionHandler (apache#7394) MINOR:fixed typo and removed outdated varilable name (apache#7402) KAFKA-8934: Create version file during build for Streams (apache#7397) KAFKA-8319: Make KafkaStreamsTest a non-integration test class (apache#7382) KAFKA-6883: Add toUpperCase support to sasl.kerberos.principal.to.local rule (KIP-309) KAFKA-8907; Return topic configs in CreateTopics response (KIP-525) (apache#7380) MINOR: Address review comments for KIP-504 authorizer changes (apache#7379) MINOR: add versioning to request and response headers (apache#7372) KAFKA-7273: Extend Connect Converter to support headers (apache#6362) MINOR: improve the Kafka RPC code generator (apache#7340) MINOR: Improve the org.apache.kafka.common.protocol code (apache#7344) KAFKA-8880: Docs on upgrade-guide (apache#7385) KAFKA-8179: do not suspend standby tasks during rebalance (apache#7321) KAFKA-8580: Compute RocksDB metrics (apache#7263) KAFKA-8880: Add overloaded function of Consumer.committed (apache#7304) HOTFIX: fix Kafka Streams upgrade note for broker backward compatibility (apache#7363) KAFKA-8848; Update system tests to use new AclAuthorizer (apache#7374) MINOR: remove unnecessary null check (apache#7299) KAFKA-6958: Overload methods for group and windowed stream to allow to name operation name using the new Named class (apache#6413) ...

add overloaded function

c2bf2ff

bbejeck approved these changes Sep 10, 2019

View reviewed changes

mjsax added the consumer label Sep 10, 2019

kamalcph approved these changes Sep 11, 2019

View reviewed changes

guozhangwang added 4 commits September 13, 2019 15:47

Merge branch 'trunk' of https://github.com/apache/kafka into K8880-ba…

da9b4e5

…tch-committed

Merge branch 'trunk' of https://github.com/apache/kafka into K8880-ba…

c3ad1d6

…tch-committed

refactor all unit tests

649fc4f

github comments

b0fe68e

remove imports

603d7b2

cpettitt-confluent reviewed Sep 16, 2019

View reviewed changes

guozhangwang mentioned this pull request Sep 16, 2019

[KAFKA-7994] Improve Stream time accuracy for restarts and rebalances #6694

Merged

3 tasks

guozhangwang added 3 commits September 16, 2019 15:49

Merge branch 'trunk' of https://github.com/apache/kafka into K8880-ba…

4168701

…tch-committed

github comments

4573451

minor refactoring

aefb5bf

guozhangwang changed the title ~~KAFKA-8880: Add overloaded function of Consumer.committed [WIP]~~ KAFKA-8880: Add overloaded function of Consumer.committed Sep 16, 2019

rebase from trunk

57f2a9a

guozhangwang commented Sep 20, 2019

View reviewed changes

replace deprecated API

2bc8bf0

bbejeck approved these changes Sep 20, 2019

View reviewed changes

guozhangwang merged commit bcc0237 into apache:trunk Sep 24, 2019

mjsax mentioned this pull request Oct 8, 2019

MINOR: unify calls to get committed offsets and metadata #7463

Merged

guozhangwang deleted the K8880-batch-committed branch April 24, 2020 23:58

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8880: Add overloaded function of Consumer.committed #7304

KAFKA-8880: Add overloaded function of Consumer.committed #7304

guozhangwang commented Sep 5, 2019 •

edited

bbejeck left a comment

bbejeck Sep 10, 2019

guozhangwang Sep 15, 2019

kamalcph left a comment

kamalcph Sep 11, 2019

kamalcph Sep 11, 2019

guozhangwang Sep 15, 2019

kamalcph Sep 11, 2019

kamalcph Sep 11, 2019

guozhangwang Sep 15, 2019

kamalcph Sep 11, 2019

guozhangwang commented Sep 15, 2019

cpettitt-confluent left a comment

cpettitt-confluent Sep 16, 2019

cpettitt-confluent Sep 16, 2019

guozhangwang Sep 16, 2019

cpettitt-confluent Sep 16, 2019

guozhangwang Sep 16, 2019

cpettitt-confluent Sep 16, 2019

cpettitt-confluent Sep 16, 2019

cpettitt-confluent Sep 16, 2019

cpettitt-confluent Sep 16, 2019

guozhangwang Sep 16, 2019

cpettitt-confluent Sep 16, 2019

cpettitt-confluent Sep 16, 2019

guozhangwang Sep 20, 2019

cpettitt-confluent Sep 20, 2019

bbejeck left a comment

bbejeck Sep 20, 2019

guozhangwang Sep 20, 2019

ConcurrencyPractitioner Jan 3, 2020

guozhangwang commented Sep 21, 2019

guozhangwang commented Sep 23, 2019

guozhangwang commented Sep 24, 2019

KAFKA-8880: Add overloaded function of Consumer.committed #7304

KAFKA-8880: Add overloaded function of Consumer.committed #7304

Conversation

guozhangwang commented Sep 5, 2019 • edited

Committer Checklist (excluded from commit message)

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kamalcph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Sep 15, 2019

cpettitt-confluent left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Sep 21, 2019

guozhangwang commented Sep 23, 2019

guozhangwang commented Sep 24, 2019

guozhangwang commented Sep 5, 2019 •

edited