Skip to content

KAFKA-8483/KAFKA-8484; Ensure safe handling of producerId resets #6883

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 6 commits into from
Jun 12, 2019
Merged

KAFKA-8483/KAFKA-8484; Ensure safe handling of producerId resets #6883

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e0f6910
KAFKA-8483; Ensure message ordering is preserved after sequence resets
hachikuji Jun 5, 2019
3eecd6b
Move completion/failure logic into TransactionManager and clean up te…
hachikuji Jun 5, 2019
349de80
Fix check for responses following producerId reset
hachikuji Jun 5, 2019
2b2e653
Need to protect batch completion as well
hachikuji Jun 5, 2019
e3727a8
Fix failing test
hachikuji Jun 5, 2019
14e75b6
Remove redundant producerId and epoch check in adjustSequenceNumbers
hachikuji Jun 12, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -558,7 +558,7 @@ private List<ProducerBatch> drainBatchesForOneNode(Cluster cluster, Node node, i
if (shouldStopDrainBatchesForPartition(first, tp))
break;

boolean isTransactional = transactionManager != null ? transactionManager.isTransactional() : false;
boolean isTransactional = transactionManager != null && transactionManager.isTransactional();
ProducerIdAndEpoch producerIdAndEpoch =
transactionManager != null ? transactionManager.producerIdAndEpoch() : null;
ProducerBatch batch = deque.pollFirst();
Expand Down
53 changes: 12 additions & 41 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.kafka.clients.producer.internals;

import java.util.ArrayList;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientRequest;
import org.apache.kafka.clients.ClientResponse;
Expand All @@ -33,11 +32,9 @@
import org.apache.kafka.common.errors.ClusterAuthorizationException;
import org.apache.kafka.common.errors.InvalidMetadataException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.InitProducerIdRequestData;
Expand All @@ -60,6 +57,7 @@
import org.slf4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
Expand Down Expand Up @@ -295,9 +293,7 @@ public void run() {
void runOnce() {
if (transactionManager != null) {
try {
if (transactionManager.shouldResetProducerStateAfterResolvingSequences())
// Check if the previous run expired batches which requires a reset of the producer state.
transactionManager.resetProducerId();
transactionManager.resetProducerIdIfNeeded();

if (!transactionManager.isTransactional()) {
// this is an idempotent producer, so make sure we have a producer id
Expand Down Expand Up @@ -694,16 +690,7 @@ private void reenqueueBatch(ProducerBatch batch, long currentTimeMs) {

private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response) {
if (transactionManager != null) {
if (transactionManager.hasProducerIdAndEpoch(batch.producerId(), batch.producerEpoch())) {
transactionManager
.maybeUpdateLastAckedSequence(batch.topicPartition, batch.baseSequence() + batch.recordCount - 1);
log.debug("ProducerId: {}; Set last ack'd sequence number for topic-partition {} to {}",
batch.producerId(),
batch.topicPartition,
transactionManager.lastAckedSequence(batch.topicPartition).orElse(-1));
}
transactionManager.updateLastAckedOffset(response, batch);
transactionManager.removeInFlightBatch(batch);
transactionManager.handleCompletedBatch(batch, response);
}

if (batch.done(response.baseOffset, response.logAppendTime, null)) {
Expand All @@ -712,36 +699,20 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons
}
}

private void failBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, RuntimeException exception,
private void failBatch(ProducerBatch batch,
ProduceResponse.PartitionResponse response,
RuntimeException exception,
boolean adjustSequenceNumbers) {
failBatch(batch, response.baseOffset, response.logAppendTime, exception, adjustSequenceNumbers);
}

private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception,
boolean adjustSequenceNumbers) {
private void failBatch(ProducerBatch batch,
long baseOffset,
long logAppendTime,
RuntimeException exception,
boolean adjustSequenceNumbers) {
if (transactionManager != null) {
if (exception instanceof OutOfOrderSequenceException
&& !transactionManager.isTransactional()
&& transactionManager.hasProducerId(batch.producerId())) {
log.error("The broker returned {} for topic-partition " +
"{} at offset {}. This indicates data loss on the broker, and should be investigated.",
exception, batch.topicPartition, baseOffset);

// Reset the transaction state since we have hit an irrecoverable exception and cannot make any guarantees
// about the previously committed message. Note that this will discard the producer id and sequence
// numbers for all existing partitions.
transactionManager.resetProducerId();
} else if (exception instanceof ClusterAuthorizationException
|| exception instanceof TransactionalIdAuthorizationException
|| exception instanceof ProducerFencedException
|| exception instanceof UnsupportedVersionException) {
transactionManager.transitionToFatalError(exception);
} else if (transactionManager.isTransactional()) {
transactionManager.transitionToAbortableError(exception);
}
transactionManager.removeInFlightBatch(batch);
if (adjustSequenceNumbers)
transactionManager.adjustSequencesDueToFailedBatch(batch);
transactionManager.handleFailedBatch(batch, exception, adjustSequenceNumbers);
}

this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount);
Expand Down
Loading