KAFKA-5886: Implement KIP-91 delivery.timeout.ms #3849
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -77,13 +77,13 @@ private enum FinalState { ABORTED, FAILED, SUCCEEDED } | |
| private boolean retry; | ||
| private boolean reopened = false; | ||
|
|
||
| public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long creationTime) { | ||
| this(tp, recordsBuilder, creationTime, false); | ||
| } | ||
|
|
||
| public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long creationTime, boolean isSplitBatch) { | ||
| this.createdMs = creationTime; | ||
| this.lastAttemptMs = creationTime; | ||
| this.recordsBuilder = recordsBuilder; | ||
| this.topicPartition = tp; | ||
| this.lastAppendTime = createdMs; | ||
|
|
@@ -158,34 +158,50 @@ public void abort(RuntimeException exception) { | |
| } | ||
|
|
||
| /** | ||
| * Finalize the state of a batch. Final state, once set, is immutable. This function may be called | ||
| * once or twice on a batch. It may be called twice if | ||
| * 1. An inflight batch expires before a response from the broker is received. The batch's final | ||
| * state is set to FAILED. But it could succeed on the broker and second time around batch.done() may | ||
| * try to set SUCCEEDED final state. | ||
| * 2. If a transaction abortion happens or if the producer is closed forcefully, the final state is | ||
| * ABORTED but again it could succeed if broker responds with a success. | ||
| * | ||
| * Attempted transitions from [FAILED | ABORTED] --> SUCCEEDED are logged. | ||
| * Attempted transitions from one failure state to the same or a different failed state are ignored. | ||
| * Attempted transitions from SUCCEEDED to the same or a failed state throw an exception. | ||
| * | ||
| * @param baseOffset The base offset of the messages assigned by the server | ||
| * @param logAppendTime The log append time or -1 if CreateTime is being used | ||
| * @param exception The exception that occurred (or null if the request was successful) | ||
| * @return true if the batch was completed successfully and false if the batch was previously aborted | ||
| */ | ||
| public boolean done(long baseOffset, long logAppendTime, RuntimeException exception) { | ||
| final FinalState tryFinalState = (exception == null) ? FinalState.SUCCEEDED : FinalState.FAILED; | ||
|
|
||
| if (tryFinalState == FinalState.SUCCEEDED) { | ||
| log.trace("Successfully produced messages to {} with base offset {}.", topicPartition, baseOffset); | ||
| } else { | ||
| log.trace("Failed to produce messages to {}.", topicPartition, exception); | ||
| } | ||
|
|
||
| if (this.finalState.compareAndSet(null, tryFinalState)) { | ||
| completeFutureAndFireCallbacks(baseOffset, logAppendTime, exception); | ||
| return true; | ||
| } | ||
|
|
||
| if (this.finalState.get() != FinalState.SUCCEEDED) { | ||
| if (tryFinalState == FinalState.SUCCEEDED) { | ||
| // Log if a previously unsuccessful batch succeeded later on. | ||
| log.debug("ProduceResponse returned {} for {} after batch had already been {}.", tryFinalState, | ||
| topicPartition, this.finalState.get()); | ||
| } | ||
| // FAILED --> ABORTED and ABORTED --> FAILED transitions are ignored. | ||
| } else { | ||
| // A SUCCESSFUL batch must not attempt another state change. | ||
| throw new IllegalStateException("A " + this.finalState.get() + " batch must not attempt another state change to " + tryFinalState); | ||
| } | ||
|
|
||
| return false; | ||
| } | ||
|
|
||
| private void completeFutureAndFireCallbacks(long baseOffset, long logAppendTime, RuntimeException exception) { | ||
|
|
@@ -300,29 +316,33 @@ public String toString() { | |
| } | ||
|
|
||
| /** | ||
| * Expire the batch if current time has passed the delivery timeout of the batch. | ||
| * The delivery timeout is measured from batch creation time instead of batch close time | ||
| * because in some cases batch close may be arbitrarily delayed. | ||
| * This methods closes this batch and sets {@code expiryErrorMessage} if the batch has timed out. | ||
| */ | ||
| boolean maybeExpire(long deliveryTimeoutMs, long now) { | ||
| if (deliveryTimeoutMs <= (now - this.createdMs)) { | ||
|
Contributor
There was a problem hiding this comment. We are using the creation time of the batch to check for expiration. That will tend to expire some records which were added to the batch after creation earlier than the delivery timeout (by as much as linger.ms). Alternatively, we could use the time that the batch was closed, which will tend to expire records later than the delivery timeout (by as much as linger.ms), but maybe expiring late is bit safer than expiring early? This is equivalent to saying that the delivery timeout excludes linger time.
Contributor
Author
There was a problem hiding this comment. Batch close may be arbitrrily delayed in some cases. See @junrao's explanation: http://search-hadoop.com/m/Kafka/uyzND1calgR1Udv2J?subj=Re+DISCUSS+KIP+91+Provide+Intuitive+User+Timeouts+in+The+Producer
Contributor
There was a problem hiding this comment. Thanks. I synced with Jun and it seems reasonable. It would help to document somewhere why we use create time. |
||
| expire(now); | ||
| return true; | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| /** | ||
| * Expire the batch. This methods closes this batch and sets {@code expiryErrorMessage}. | ||
| */ | ||
| void expire(long now) { | ||
| expiryErrorMessage = (now - this.createdMs) + " ms has passed since batch creation"; | ||
| abortRecordAppends(); | ||
|
Contributor
There was a problem hiding this comment. Checking my understanding. With this change, it should no longer be possible to expire a batch before linger.ms has completed and the batch has been closed. If so, do we still need the logic to abort appends on expiration? (It might be safer to have it anyway, just checking if it is still needed for correctness)
Contributor
There was a problem hiding this comment. I think the answer to this question is that it is possible to expire while the batch is still being built because closing the batch can be arbitrarily delayed by inflight fetches. |
||
| } | ||
|
|
||
| public FinalState finalState() { | ||
| return this.finalState.get(); | ||
| } | ||
|
|
||
| /** | ||
| * If {@link #maybeExpire(long, long)} returned true, the sender will fail the batch with | ||
| * the exception returned by this method. | ||
| * @return An exception indicating the batch expired. | ||
| */ | ||
|
|
@@ -468,4 +488,9 @@ public boolean sequenceHasBeenReset() { | |
| return reopened; | ||
| } | ||
|
|
||
| boolean soonToExpire(long deliveryTimeoutMs, long now, int requestTimeoutMs) { | ||
| // We don't add anything in deliveryTimeoutMs because it may overflow | ||
| // when deliveryTimeoutMs==Long.MAX_VALUE | ||
| return deliveryTimeoutMs <= now + requestTimeoutMs - createdMs; | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
verifyAndGetDeliveryTimeout()?