KAFKA-8421: Still return data during rebalance #7312
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -1231,9 +1231,11 @@ private ConsumerRecords<K, V> poll(final Timer timer, final boolean includeMetad | |||||
| client.maybeTriggerWakeup(); | ||||||
|
|
||||||
| if (includeMetadataInTimeout) { | ||||||
| // try to update assignment metadata BUT do not need to block on the timer, | ||||||
| // since even if we are 1) in the middle of a rebalance or 2) have partitions | ||||||
| // with unknown starting positions we may still want to return some data | ||||||
| // as long as there are some partitions fetchable | ||||||
| updateAssignmentMetadataIfNeeded(timer.remainingMs() > 0 ? time.timer(1L) : timer); | ||||||
| } else { | ||||||
| while (!updateAssignmentMetadataIfNeeded(time.timer(Long.MAX_VALUE))) { | ||||||
| log.warn("Still waiting for metadata"); | ||||||
|
|
@@ -1307,12 +1309,6 @@ private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(Timer tim | |||||
| }); | ||||||
| timer.update(pollTimer.currentTimeMs()); | ||||||
|
|
||||||
|
Contributor
There was a problem hiding this comment. Why is this dropped? Are you thinking it's just redundant?
Member
There was a problem hiding this comment. Isn't this the "main point" of this PR -- we don't want to return nothing just because a rebalance is needed or pending
Contributor
There was a problem hiding this comment. Oh, yeah... that makes sense. |
||||||
| return fetcher.fetchedRecords(); | ||||||
| } | ||||||
|
|
||||||
|
|
@@ -1334,6 +1330,9 @@ private Map<TopicPartition, List<ConsumerRecord<K, V>>> pollForFetches(Timer tim | |||||
| * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. | ||||||
| * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, | ||||||
| * or if there is an active group with the same groupId which is using group management. | ||||||
| * @throws org.apache.kafka.clients.consumer.RetriableCommitFailedException if the commit failed but can be retriable. | ||||||
|
Contributor
There was a problem hiding this comment. Just to clarify:
Member
There was a problem hiding this comment. "Can be retriable" does seem to imply that in some circumstances it can be retried, while in others it can't be retried at all. "Can be retried" implies it's up to the user whether they want to retry
Contributor
Author
There was a problem hiding this comment. I think |
||||||
| * This can occur if, e.g. consumer instance is in the middle of a rebalance. In such cases | ||||||
| * commit could be retried after the rebalance is completed with the {@link #poll(Duration)} call. | ||||||
| * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this | ||||||
| * function is called | ||||||
| * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while | ||||||
|
|
@@ -1369,6 +1368,9 @@ public void commitSync() { | |||||
| * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. | ||||||
| * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, | ||||||
| * or if there is an active group with the same groupId which is using group management. | ||||||
| * @throws org.apache.kafka.clients.consumer.RetriableCommitFailedException if the commit failed but can be retriable. | ||||||
| * This can occur if, e.g. consumer instance is in the middle of a rebalance. In such cases | ||||||
| * commit could be retried after the rebalance is completed with the {@link #poll(Duration)} call. | ||||||
| * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this | ||||||
| * function is called | ||||||
| * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while | ||||||
|
|
@@ -1402,7 +1404,8 @@ public void commitSync(Duration timeout) { | |||||
| * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every | ||||||
| * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API | ||||||
| * should not be used. The committed offset should be the next message your application will consume, | ||||||
| * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, | ||||||
| * then the committed offsets must belong to the currently auto-assigned partitions. | ||||||
| * <p> | ||||||
| * This is a synchronous commits and will block until either the commit succeeds or an unrecoverable error is | ||||||
| * encountered (in which case it is thrown to the caller), or the timeout specified by {@code default.api.timeout.ms} expires | ||||||
|
|
@@ -1415,6 +1418,9 @@ public void commitSync(Duration timeout) { | |||||
| * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. | ||||||
|
Member
There was a problem hiding this comment. nit: Can we make this more specific, or maybe provide examples? eg this happens if you try to commit offsets for partitions you don't own, or you fell out of the group |
||||||
| * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, | ||||||
| * or if there is an active group with the same groupId which is using group management. | ||||||
| * @throws org.apache.kafka.clients.consumer.RetriableCommitFailedException if the commit failed but can be retriable. | ||||||
| * This can occur if, e.g. consumer instance is in the middle of a rebalance. In such cases | ||||||
|
Member
There was a problem hiding this comment. nit: If a user tries to commit during an eager rebalance, the |
||||||
| * commit could be retried after the rebalance is completed with the {@link #poll(Duration)} call. | ||||||
| * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this | ||||||
| * function is called | ||||||
| * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while | ||||||
|
|
@@ -1435,43 +1441,48 @@ public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets) { | |||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Commit the specified offsets for the specified list of topics and partitions. | ||||||
| * <p> | ||||||
| * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every | ||||||
| * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API | ||||||
| * should not be used. The committed offset should be the next message your application will consume, | ||||||
| * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, | ||||||
| * then the committed offsets must belong to the currently auto-assigned partitions. | ||||||
| * <p> | ||||||
| * This is a synchronous commits and will block until either the commit succeeds, an unrecoverable error is | ||||||
|
Contributor
There was a problem hiding this comment.
Suggested change
|
||||||
| * encountered (in which case it is thrown to the caller), or the timeout expires. | ||||||
| * <p> | ||||||
| * Note that asynchronous offset commits sent previously with the {@link #commitAsync(OffsetCommitCallback)} | ||||||
| * (or similar) are guaranteed to have their callbacks invoked prior to completion of this method. | ||||||
| * | ||||||
| * @param offsets A map of offsets by partition with associated metadata | ||||||
| * @param timeout The maximum amount of time to await completion of the offset commit | ||||||
| * @throws org.apache.kafka.clients.consumer.CommitFailedException if the commit failed and cannot be retried. | ||||||
| * This can only occur if you are using automatic group management with {@link #subscribe(Collection)}, | ||||||
| * or if there is an active group with the same groupId which is using group management. | ||||||
| * @throws org.apache.kafka.clients.consumer.RetriableCommitFailedException if the commit failed but can be retriable. | ||||||
| * This can occur if, e.g. consumer instance is in the middle of a rebalance. In such cases | ||||||
| * commit could be retried after the rebalance is completed with the {@link #poll(Duration)} call. | ||||||
| * @throws org.apache.kafka.common.errors.WakeupException if {@link #wakeup()} is called before or while this | ||||||
| * function is called | ||||||
| * @throws org.apache.kafka.common.errors.InterruptException if the calling thread is interrupted before or while | ||||||
| * this function is called | ||||||
| * @throws org.apache.kafka.common.errors.AuthenticationException if authentication fails. See the exception for more details | ||||||
| * @throws org.apache.kafka.common.errors.AuthorizationException if not authorized to the topic or to the | ||||||
| * configured groupId. See the exception for more details | ||||||
| * @throws java.lang.IllegalArgumentException if the committed offset is negative | ||||||
| * @throws org.apache.kafka.common.KafkaException for any other unrecoverable errors (e.g. if offset metadata | ||||||
| * is too large or if the topic does not exist). | ||||||
| * @throws org.apache.kafka.common.errors.TimeoutException if the timeout expires before successful completion | ||||||
| * of the offset commit | ||||||
| * @throws org.apache.kafka.common.errors.FencedInstanceIdException if this consumer instance gets fenced by broker. | ||||||
| */ | ||||||
| @Override | ||||||
| public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsets, final Duration timeout) { | ||||||
| acquireAndEnsureOpen(); | ||||||
| try { | ||||||
| maybeThrowInvalidGroupIdException(); | ||||||
| maybeThrowIfCommitOffsetsNotOwned(offsets); | ||||||
| offsets.forEach(this::updateLastSeenEpochIfNewer); | ||||||
| if (!coordinator.commitOffsetsSync(new HashMap<>(offsets), time.timer(timeout))) { | ||||||
| throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before successfully " + | ||||||
|
|
@@ -1521,7 +1532,8 @@ public void commitAsync(OffsetCommitCallback callback) { | |||||
| * This commits offsets to Kafka. The offsets committed using this API will be used on the first fetch after every | ||||||
| * rebalance and also on startup. As such, if you need to store offsets in anything other than Kafka, this API | ||||||
| * should not be used. The committed offset should be the next message your application will consume, | ||||||
| * i.e. lastProcessedMessageOffset + 1. If automatic group management with {@link #subscribe(Collection)} is used, | ||||||
| * then the committed offsets must belong to the currently auto-assigned partitions. | ||||||
| * <p> | ||||||
| * This is an asynchronous call and will not block. Any errors encountered are either passed to the callback | ||||||
| * (if provided) or discarded. | ||||||
|
|
@@ -1541,14 +1553,29 @@ public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, Of | |||||
| acquireAndEnsureOpen(); | ||||||
| try { | ||||||
| maybeThrowInvalidGroupIdException(); | ||||||
| maybeThrowIfCommitOffsetsNotOwned(offsets); | ||||||
|
Contributor
There was a problem hiding this comment. Did you consider the compatibility of this change? We probably should have been more restrictive from the beginning, but is it a risk to change now? An alternative might be to just log a warning.
Contributor
Author
There was a problem hiding this comment. Originally I thought about this as a bug fix since we only check generation id at the broker-side, but now as I examine the javadoc again, we never state that for these two functions users should only commit on auto-assigned partitions -- in fact, if that's the case then they likely should just use the other two overloads unless they do not want the current position to be committed -- like in Streams, if some records get deserialization errors. So after a second thought I'm happy to take it out of this PR and still allow users to do so (perhaps I'd just make it clear that, if you use these two functions we would NOT check it so you can actually commit whatever you want and calling these two functions with your own caution). WYDT @hachikuji ?
Contributor
There was a problem hiding this comment. It seems safer that way, but it does feel a little unsatisfying to be left in this state forever since we probably should have always had the validation. Perhaps we could file a JIRA and add the validation in 3.0 or something?
Contributor
Author
There was a problem hiding this comment. Now I actually feel if it is by-design to allow user this flexibility with auto-assignment but with their own caution? If we always want to restrict to assigned partitions then most likely they would not need these two overloads. |
||||||
| log.debug("Committing offsets: {}", offsets); | ||||||
| offsets.forEach(this::updateLastSeenEpochIfNewer); | ||||||
| coordinator.commitOffsetsAsync(new HashMap<>(offsets), callback); | ||||||
| } catch (CommitFailedException e) { | ||||||
| log.error("Failed to commit offsets asynchronously because they do not belong to dynamically assigned partitions"); | ||||||
|
Member
There was a problem hiding this comment. Just wondering, does this mean we did not used to check if all the offsets actually belonged to auto-assigned partitions? Or did we just straight throw the
Contributor
Author
There was a problem hiding this comment. I think this is not a behavior change: before this, the broker would check the commit offset request (but only on the generation id, as the coordinator did not keep trace who-owns-what), and if failed it will return the error code which would be translated to CommitFailed on the callback still. With this we will set the same error on the callback, but much earlier and skip the offset commit request.
Contributor
There was a problem hiding this comment. Do we still need this |
||||||
| callback.onComplete(offsets, e); | ||||||
| } finally { | ||||||
| release(); | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| private void maybeThrowIfCommitOffsetsNotOwned(final Map<TopicPartition, OffsetAndMetadata> offsets) { | ||||||
|
Contributor
There was a problem hiding this comment. nit: The callers of this iterate the committed offsets twice. Once in order to check partition ownership and then a second time to invoke
Contributor
Author
There was a problem hiding this comment. Yes, if we are not gonna remove this (see other comment above) I can do this. |
||||||
| if (subscriptions.partitionsAutoAssigned()) { | ||||||
| final Set<TopicPartition> partitions = assignment(); | ||||||
| for (final TopicPartition tp : offsets.keySet()) { | ||||||
| if (!partitions.contains(tp)) | ||||||
| throw new CommitFailedException("Cannot commit offset of topic partition " + tp + | ||||||
| " since the partition was not dynamically assigned to this consumer"); | ||||||
| } | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| /** | ||||||
| * Overrides the fetch offsets that the consumer will use on the next {@link #poll(Duration) poll(timeout)}. If this API | ||||||
| * is invoked for the same partition more than once, the latest offset will be used on the next poll(). Note that | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -111,7 +111,7 @@ | |
| public abstract class AbstractCoordinator implements Closeable { | ||
| public static final String HEARTBEAT_THREAD_PREFIX = "kafka-coordinator-heartbeat-thread"; | ||
|
|
||
| protected enum MemberState { | ||
| UNJOINED, // the client is not part of a group | ||
| REBALANCING, // the client has begun rebalancing | ||
| STABLE, // the client has joined and is sending heartbeats | ||
|
|
@@ -130,11 +130,11 @@ private enum MemberState { | |
| private MemberState state = MemberState.UNJOINED; | ||
| private HeartbeatThread heartbeatThread = null; | ||
| private RequestFuture<ByteBuffer> joinFuture = null; | ||
| private RequestFuture<Void> findCoordinatorFuture = null; | ||
| private Generation generation = Generation.NO_GENERATION; | ||
| private long lastRebalanceStartMs = -1L; | ||
| private long lastRebalanceEndMs = -1L; | ||
|
|
||
|
|
||
| /** | ||
| * Initialize the coordination manager. | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I've been trying to follow this. Looking into
ConsumerCoordinator, I think I see what you're talking about, there are several opportunities to return early if the timer expires. But is1msenough to guarantee that we'll actually get through to where we fetch the data? Is it possible to just modify the conditionals farther down to make sure that we can always make progress even when the timer is set to 0ms (i.e., the request is fully asynchronous). To me, a timeout of 0ms doesn't mean that the operation should take no time at all, just that it shouldn't block. It seems "within bounds" to still perform any operation we need in order to guarantee we make progress.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, the original problem is that with
timeout(0), we cannot receive the join-group response and send the sync-group request in a single call as it is two round-trips; so I made a 1ms here trying to proceed faster instead requesting users to callpoll()repeatedly. But now I realized that while fixing 6df058e theclient.transmitSends()actually worked around it so now we can indeed call with 0m always.I'll change it back to always call 0ms now.