diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java index 18248bb782d9f..b630d611dc235 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java @@ -365,7 +365,7 @@ private KafkaProducer(ProducerConfig config, Serializer keySerializer, Serial this.requestTimeoutMs = config.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG); this.transactionManager = configureTransactionState(config, logContext, log); int retries = configureRetries(config, transactionManager != null, log); - int maxInflightRequests = configureInflightRequests(config, transactionManager != null, log); + int maxInflightRequests = configureInflightRequests(config, transactionManager != null); short acks = configureAcks(config, transactionManager != null, log); this.apiVersions = new ApiVersions(); @@ -481,18 +481,10 @@ private static int configureRetries(ProducerConfig config, boolean idempotenceEn return config.getInt(ProducerConfig.RETRIES_CONFIG); } - private static int configureInflightRequests(ProducerConfig config, boolean idempotenceEnabled, Logger log) { - boolean userConfiguredInflights = false; - if (config.originals().containsKey(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { - userConfiguredInflights = true; - } - if (idempotenceEnabled && !userConfiguredInflights) { - log.info("Overriding the default {} to 1 since idempontence is enabled.", ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); - return 1; - } - if (idempotenceEnabled && config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION) != 1) { - throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to 1 in order" + - "to use the idempotent producer. Otherwise we cannot guarantee idempotence."); + private static int configureInflightRequests(ProducerConfig config, boolean idempotenceEnabled) { + if (idempotenceEnabled && 5 < config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { + throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" + + " to use the idempotent producer."); } return config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); } diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java index ee7d21a5df198..93c843b9163a1 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java @@ -75,6 +75,7 @@ private enum FinalState { ABORTED, FAILED, SUCCEEDED } private long drainedMs; private String expiryErrorMessage; private boolean retry; + private boolean reopened = false; public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long now) { this(tp, recordsBuilder, now, false); @@ -249,6 +250,15 @@ public Deque split(int splitBatchSize) { produceFuture.set(ProduceResponse.INVALID_OFFSET, NO_TIMESTAMP, new RecordBatchTooLargeException()); produceFuture.done(); + + if (hasSequence()) { + int sequence = baseSequence(); + ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId(), producerEpoch()); + for (ProducerBatch newBatch : batches) { + newBatch.setProducerState(producerIdAndEpoch, sequence, isTransactional()); + sequence += newBatch.recordCount; + } + } return batches; } @@ -375,8 +385,12 @@ public boolean isFull() { } public void setProducerState(ProducerIdAndEpoch producerIdAndEpoch, int baseSequence, boolean isTransactional) { - recordsBuilder.setProducerState(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, - baseSequence, isTransactional); + recordsBuilder.setProducerState(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, baseSequence, isTransactional); + } + + public void resetProducerState(ProducerIdAndEpoch producerIdAndEpoch, int baseSequence, boolean isTransactional) { + reopened = true; + recordsBuilder.reopenAndRewriteProducerState(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, baseSequence, isTransactional); } /** @@ -394,6 +408,7 @@ public void close() { recordsBuilder.compressionType(), (float) recordsBuilder.compressionRatio()); } + reopened = false; } /** @@ -434,4 +449,21 @@ public long producerId() { public short producerEpoch() { return recordsBuilder.producerEpoch(); } + + public int baseSequence() { + return recordsBuilder.baseSequence(); + } + + public boolean hasSequence() { + return baseSequence() != RecordBatch.NO_SEQUENCE; + } + + public boolean isTransactional() { + return recordsBuilder.isTransactional(); + } + + public boolean sequenceHasBeenReset() { + return reopened; + } + } diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index 38b5e517fcf84..f3cb4faf8573b 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java @@ -251,7 +251,8 @@ private MemoryRecordsBuilder recordsBuilder(ByteBuffer buffer, byte maxUsableMag * and memory records built) in one of the following cases (whichever comes first): right before send, * if it is expired, or when the producer is closed. */ - private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, Deque deque) { + private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, + Callback callback, Deque deque) { ProducerBatch last = deque.peekLast(); if (last != null) { FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, time.milliseconds()); @@ -259,7 +260,6 @@ private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, H last.closeForRecordAppends(); else return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false); - } return null; } @@ -309,7 +309,10 @@ public void reenqueue(ProducerBatch batch, long now) { batch.reenqueued(now); Deque deque = getOrCreateDeque(batch.topicPartition); synchronized (deque) { - deque.addFirst(batch); + if (transactionManager != null) + insertInSequenceOrder(deque, batch); + else + deque.addFirst(batch); } } @@ -331,12 +334,71 @@ public int splitAndReenqueue(ProducerBatch bigBatch) { incomplete.add(batch); // We treat the newly split batches as if they are not even tried. synchronized (partitionDequeue) { - partitionDequeue.addFirst(batch); + if (transactionManager != null) { + // We should track the newly created batches since they already have assigned sequences. + transactionManager.addInFlightBatch(batch); + insertInSequenceOrder(partitionDequeue, batch); + } else { + partitionDequeue.addFirst(batch); + } } } return numSplitBatches; } + // The deque for the partition may have to be reordered in situations where leadership changes in between + // batch drains. Since the requests are on different connections, we no longer have any guarantees about ordering + // of the responses. Hence we will have to check if there is anything out of order and ensure the batch is queued + // in the correct sequence order. + // + // Note that this assumes that all the batches in the queue which have an assigned sequence also have the current + // producer id. We will not attempt to reorder messages if the producer id has changed. + + private void insertInSequenceOrder(Deque deque, ProducerBatch batch) { + // When we are requeing and have enabled idempotence, the reenqueued batch must always have a sequence. + if (batch.baseSequence() == RecordBatch.NO_SEQUENCE) + throw new IllegalStateException("Trying to reenqueue a batch which doesn't have a sequence even " + + "though idempotence is enabled."); + + if (transactionManager.nextBatchBySequence(batch.topicPartition) == null) + throw new IllegalStateException("We are reenqueueing a batch which is not tracked as part of the in flight " + + "requests. batch.topicPartition: " + batch.topicPartition + "; batch.baseSequence: " + batch.baseSequence()); + + // If there are no inflight batches being tracked by the transaction manager, it means that the producer + // id must have changed and the batches being re enqueued are from the old producer id. In this case + // we don't try to ensure ordering amongst them. They will eventually fail with an OutOfOrderSequence, + // or they will succeed. + if (batch.baseSequence() != transactionManager.nextBatchBySequence(batch.topicPartition).baseSequence()) { + // The incoming batch can't be inserted at the front of the queue without violating the sequence ordering. + // This means that the incoming batch should be placed somewhere further back. + // We need to find the right place for the incoming batch and insert it there. + // We will only enter this branch if we have multiple inflights sent to different brokers, perhaps + // because a leadership change occurred in between the drains. In this scenario, responses can come + // back out of order, requiring us to re order the batches ourselves rather than relying on the + // implicit ordering guarantees of the network client which are only on a per connection basis. + + List orderedBatches = new ArrayList<>(); + while (deque.peekFirst() != null && deque.peekFirst().hasSequence() && deque.peekFirst().baseSequence() < batch.baseSequence()) + orderedBatches.add(deque.pollFirst()); + + log.debug("Reordered incoming batch with sequence {} for partition {}. It was placed in the queue at " + + "position {}", batch.baseSequence(), batch.topicPartition, orderedBatches.size()); + // Either we have reached a point where there are batches without a sequence (ie. never been drained + // and are hence in order by default), or the batch at the front of the queue has a sequence greater + // than the incoming batch. This is the right place to add the incoming batch. + deque.addFirst(batch); + + // Now we have to re insert the previously queued batches in the right order. + for (int i = orderedBatches.size() - 1; i >= 0; --i) { + deque.addFirst(orderedBatches.get(i)); + } + + // At this point, the incoming batch has been queued in the correct place according to its sequence. + } else { + deque.addFirst(batch); + } + } + /** * Get a list of nodes whose partitions are ready to be sent, and the earliest time at which any non-sendable * partition will be ready; Also return the flag for whether there are any unknown leaders for the accumulated @@ -469,20 +531,42 @@ public Map> drain(Cluster cluster, break; isTransactional = transactionManager.isTransactional(); + + if (!first.hasSequence() && transactionManager.hasUnresolvedSequence(first.topicPartition)) + // Don't drain any new batches while the state of previous sequence numbers + // is unknown. The previous batches would be unknown if they were aborted + // on the client after being sent to the broker at least once. + break; + + if (first.hasSequence() + && first.baseSequence() != transactionManager.nextBatchBySequence(first.topicPartition).baseSequence()) + // If the queued batch already has an assigned sequence, then it is being + // retried. In this case, we wait until the next immediate batch is ready + // and drain that. We only move on when the next in line batch is complete (either successfully + // or due to a fatal broker error). This effectively reduces our + // in flight request count to 1. + break; } ProducerBatch batch = deque.pollFirst(); - if (producerIdAndEpoch != null && !batch.inRetry()) { - // If the batch is in retry, then we should not change the producer id and + if (producerIdAndEpoch != null && !batch.hasSequence()) { + // If the batch already has an assigned sequence, then we should not change the producer id and // sequence number, since this may introduce duplicates. In particular, // the previous attempt may actually have been accepted, and if we change // the producer id and sequence here, this attempt will also be accepted, // causing a duplicate. - int sequenceNumber = transactionManager.sequenceNumber(batch.topicPartition); - log.debug("Assigning sequence number {} from producer {} to dequeued " + - "batch from partition {} bound for {}.", - sequenceNumber, producerIdAndEpoch, batch.topicPartition, node); - batch.setProducerState(producerIdAndEpoch, sequenceNumber, isTransactional); + // + // Additionally, we update the next sequence number bound for the partition, + // and also have the transaction manager track the batch so as to ensure + // that sequence ordering is maintained even if we receive out of order + // responses. + batch.setProducerState(producerIdAndEpoch, transactionManager.sequenceNumber(batch.topicPartition), isTransactional); + transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount); + log.debug("Assigned producerId {} and producerEpoch {} to batch with base sequence " + + "{} being sent to partition {}", producerIdAndEpoch.producerId, + producerIdAndEpoch.epoch, batch.baseSequence(), tp); + + transactionManager.addInFlightBatch(batch); } batch.close(); size += batch.records().sizeInBytes(); @@ -634,7 +718,7 @@ void abortUndrainedBatches(RuntimeException reason) { Deque dq = getDeque(batch.topicPartition); boolean aborted = false; synchronized (dq) { - if (!batch.isClosed()) { + if ((transactionManager != null && !batch.hasSequence()) || (transactionManager == null && !batch.isClosed())) { aborted = true; batch.abortRecordAppends(); dq.remove(batch); diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java index 8da411c751a07..2da47a8430a17 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java @@ -200,10 +200,17 @@ public void run() { */ void run(long now) { if (transactionManager != null) { + if (transactionManager.shouldResetProducerStateAfterResolvingSequences()) + // Check if the previous run expired batches which requires a reset of the producer state. + transactionManager.resetProducerId(); + if (!transactionManager.isTransactional()) { // this is an idempotent producer, so make sure we have a producer id maybeWaitForProducerId(); - } else if (transactionManager.hasInFlightRequest() || maybeSendTransactionalRequest(now)) { + } else if (transactionManager.hasUnresolvedSequences() && !transactionManager.hasFatalError()) { + transactionManager.transitionToFatalError(new KafkaException("The client hasn't received acknowledgment for " + + "some previously sent messages and can no longer retry them. It isn't safe to continue.")); + } else if (transactionManager.hasInFlightTransactionalRequest() || maybeSendTransactionalRequest(now)) { // as long as there are outstanding transactional requests, we simply wait for them to return client.poll(retryBackoffMs, now); return; @@ -228,6 +235,7 @@ void run(long now) { private long sendProducerData(long now) { Cluster cluster = metadata.fetch(); + // get the list of partitions with data ready to send RecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now); @@ -264,23 +272,17 @@ private long sendProducerData(long now) { } List expiredBatches = this.accumulator.expiredBatches(this.requestTimeout, now); - boolean needsTransactionStateReset = false; // Reset the producer id if an expired batch has previously been sent to the broker. Also update the metrics // for expired batches. see the documentation of @TransactionState.resetProducerId to understand why // we need to reset the producer id here. if (!expiredBatches.isEmpty()) log.trace("Expired {} batches in accumulator", expiredBatches.size()); for (ProducerBatch expiredBatch : expiredBatches) { - failBatch(expiredBatch, -1, NO_TIMESTAMP, expiredBatch.timeoutException()); + failBatch(expiredBatch, -1, NO_TIMESTAMP, expiredBatch.timeoutException(), false); if (transactionManager != null && expiredBatch.inRetry()) { - needsTransactionStateReset = true; + // This ensures that no new batches are drained until the current in flight batches are fully resolved. + transactionManager.markSequenceUnresolved(expiredBatch.topicPartition); } - this.sensors.recordErrors(expiredBatch.topicPartition.topic(), expiredBatch.recordCount); - } - - if (needsTransactionStateReset) { - transactionManager.resetProducerId(); - return 0; } sensors.updateProduceRequestMetrics(batches); @@ -345,7 +347,7 @@ private boolean maybeSendTransactionalRequest(long now) { ClientRequest clientRequest = client.newClientRequest(targetNode.idString(), requestBuilder, now, true, nextRequestHandler); - transactionManager.setInFlightRequestCorrelationId(clientRequest.correlationId()); + transactionManager.setInFlightTransactionalRequestCorrelationId(clientRequest.correlationId()); log.debug("Sending transactional request {} to node {}", requestBuilder, targetNode); client.send(clientRequest, now); @@ -422,6 +424,7 @@ private void maybeWaitForProducerId() { ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch( initProducerIdResponse.producerId(), initProducerIdResponse.epoch()); transactionManager.setProducerIdAndEpoch(producerIdAndEpoch); + return; } else if (error.exception() instanceof RetriableException) { log.debug("Retriable error from InitProducerId response", error.message()); } else { @@ -492,6 +495,7 @@ private void handleProduceResponse(ClientResponse response, Map 1 && (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 || batch.isCompressed())) { // If the batch is too large, we split the batch and send the split batches again. We do not decrement @@ -501,6 +505,8 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons batch.topicPartition, this.retries - batch.attempts(), error); + if (transactionManager != null) + transactionManager.removeInFlightBatch(batch); this.accumulator.splitAndReenqueue(batch); this.accumulator.deallocate(batch); this.sensors.recordBatchSplit(); @@ -517,14 +523,20 @@ private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionRespons // If idempotence is enabled only retry the request if the current producer id is the same as // the producer id of the batch. log.debug("Retrying batch to topic-partition {}. Sequence number : {}", batch.topicPartition, - transactionManager.sequenceNumber(batch.topicPartition)); + batch.baseSequence()); reenqueueBatch(batch, now); } else { failBatch(batch, response, new OutOfOrderSequenceException("Attempted to retry sending a " + "batch but the producer id changed from " + batch.producerId() + " to " + - transactionManager.producerIdAndEpoch().producerId + " in the mean time. This batch will be dropped.")); - this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount); + transactionManager.producerIdAndEpoch().producerId + " in the mean time. This batch will be dropped."), false); } + } else if (error == Errors.DUPLICATE_SEQUENCE_NUMBER) { + // If we have received a duplicate sequence error, it means that the sequence number has advanced beyond + // the sequence of the current batch, and we haven't retained batch metadata on the broker to return + // the correct offset and timestamp. + // + // The only thing we can do is to return success to the user and not return a valid offset and timestamp. + completeBatch(batch, response); } else { final RuntimeException exception; if (error == Errors.TOPIC_AUTHORIZATION_FAILED) @@ -533,9 +545,10 @@ else if (error == Errors.CLUSTER_AUTHORIZATION_FAILED) exception = new ClusterAuthorizationException("The producer is not authorized to do idempotent sends"); else exception = error.exception(); - // tell the user the result of their request - failBatch(batch, response, exception); - this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount); + // tell the user the result of their request. We only adjust sequence numbers if the batch didn't exhaust + // its retries -- if it did, we don't know whether the sequence number was accepted or not, and + // thus it is not safe to reassign the sequence. + failBatch(batch, response, exception, batch.attempts() < this.retries); } if (error.exception() instanceof InvalidMetadataException) { if (error.exception() instanceof UnknownTopicOrPartitionException) @@ -559,21 +572,24 @@ private void reenqueueBatch(ProducerBatch batch, long currentTimeMs) { } private void completeBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response) { - if (transactionManager != null && transactionManager.hasProducerIdAndEpoch(batch.producerId(), batch.producerEpoch())) { - transactionManager.incrementSequenceNumber(batch.topicPartition, batch.recordCount); - log.debug("Incremented sequence number for topic-partition {} to {}", batch.topicPartition, - transactionManager.sequenceNumber(batch.topicPartition)); + 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)); + } + transactionManager.removeInFlightBatch(batch); } batch.done(response.baseOffset, response.logAppendTime, null); this.accumulator.deallocate(batch); } - private void failBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, RuntimeException exception) { - failBatch(batch, response.baseOffset, response.logAppendTime, 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) { + private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception, boolean adjustSequenceNumbers) { if (transactionManager != null) { if (exception instanceof OutOfOrderSequenceException && !transactionManager.isTransactional() @@ -594,16 +610,26 @@ private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, } else if (transactionManager.isTransactional()) { transactionManager.transitionToAbortableError(exception); } + transactionManager.removeInFlightBatch(batch); + if (adjustSequenceNumbers) + transactionManager.adjustSequencesDueToFailedBatch(batch); } + + this.sensors.recordErrors(batch.topicPartition.topic(), batch.recordCount); batch.done(baseOffset, logAppendTime, exception); this.accumulator.deallocate(batch); } /** - * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed + * We can retry a send if the error is transient and the number of attempts taken is fewer than the maximum allowed. + * We can also retry OutOfOrderSequence exceptions for future batches, since if the first batch has failed, the future + * batches are certain to fail with an OutOfOrderSequence exception. */ private boolean canRetry(ProducerBatch batch, Errors error) { - return batch.attempts() < this.retries && error.exception() instanceof RetriableException; + return batch.attempts() < this.retries && + ((error.exception() instanceof RetriableException) || + (error.exception() instanceof OutOfOrderSequenceException + && transactionManager.canRetryOutOfOrderSequenceException(batch))); } /** diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java index 05d943c8fb0a3..b2387a0a74e10 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java @@ -65,7 +65,27 @@ public class TransactionManager { private final String transactionalId; private final int transactionTimeoutMs; - private final Map sequenceNumbers; + // The base sequence of the next batch bound for a given partition. + private final Map nextSequence; + + // The sequence of the last record of the last ack'd batch from the given partition. When there are no + // in flight requests for a partition, the lastAckedSequence(topicPartition) == nextSequence(topicPartition) - 1. + private final Map lastAckedSequence; + + // If a batch bound for a partition expired locally after being sent at least once, the partition has is considered + // to have an unresolved state. We keep track fo such partitions here, and cannot assign any more sequence numbers + // for this partition until the unresolved state gets cleared. This may happen if other inflight batches returned + // successfully (indicating that the expired batch actually made it to the broker). If we don't get any successful + // responses for the partition once the inflight request count falls to zero, we reset the producer id and + // consequently clear this data structure as well. + private final Set partitionsWithUnresolvedSequences; + + // Keep track of the in flight batches bound for a partition, ordered by sequence. This helps us to ensure that + // we continue to order batches by the sequence numbers even when the responses come back out of order during + // leader failover. We add a batch to the queue when it is drained, and remove it when the batch completes + // (either successfully or through a fatal failure). + private final Map> inflightBatchesBySequence; + private final PriorityQueue pendingRequests; private final Set newPartitionsInTransaction; private final Set pendingPartitionsInTransaction; @@ -142,7 +162,8 @@ private enum Priority { public TransactionManager(LogContext logContext, String transactionalId, int transactionTimeoutMs, long retryBackoffMs) { this.producerIdAndEpoch = new ProducerIdAndEpoch(NO_PRODUCER_ID, NO_PRODUCER_EPOCH); - this.sequenceNumbers = new HashMap<>(); + this.nextSequence = new HashMap<>(); + this.lastAckedSequence = new HashMap<>(); this.transactionalId = transactionalId; this.log = logContext.logger(TransactionManager.class); this.transactionTimeoutMs = transactionTimeoutMs; @@ -159,6 +180,9 @@ public int compare(TxnRequestHandler o1, TxnRequestHandler o2) { } }); + this.partitionsWithUnresolvedSequences = new HashSet<>(); + this.inflightBatchesBySequence = new HashMap<>(); + this.retryBackoffMs = retryBackoffMs; } @@ -170,7 +194,7 @@ public synchronized TransactionalRequestResult initializeTransactions() { ensureTransactional(); transitionTo(State.INITIALIZING); setProducerIdAndEpoch(ProducerIdAndEpoch.NONE); - this.sequenceNumbers.clear(); + this.nextSequence.clear(); InitProducerIdRequest.Builder builder = new InitProducerIdRequest.Builder(transactionalId, transactionTimeoutMs); InitProducerIdHandler handler = new InitProducerIdHandler(builder); enqueueRequest(handler); @@ -362,28 +386,157 @@ synchronized void resetProducerId() { throw new IllegalStateException("Cannot reset producer state for a transactional producer. " + "You must either abort the ongoing transaction or reinitialize the transactional producer instead"); setProducerIdAndEpoch(ProducerIdAndEpoch.NONE); - this.sequenceNumbers.clear(); + this.nextSequence.clear(); + this.lastAckedSequence.clear(); + this.inflightBatchesBySequence.clear(); + this.partitionsWithUnresolvedSequences.clear(); } /** * Returns the next sequence number to be written to the given TopicPartition. */ synchronized Integer sequenceNumber(TopicPartition topicPartition) { - Integer currentSequenceNumber = sequenceNumbers.get(topicPartition); + Integer currentSequenceNumber = nextSequence.get(topicPartition); if (currentSequenceNumber == null) { currentSequenceNumber = 0; - sequenceNumbers.put(topicPartition, currentSequenceNumber); + nextSequence.put(topicPartition, currentSequenceNumber); } return currentSequenceNumber; } synchronized void incrementSequenceNumber(TopicPartition topicPartition, int increment) { - Integer currentSequenceNumber = sequenceNumbers.get(topicPartition); + Integer currentSequenceNumber = nextSequence.get(topicPartition); if (currentSequenceNumber == null) throw new IllegalStateException("Attempt to increment sequence number for a partition with no current sequence."); currentSequenceNumber += increment; - sequenceNumbers.put(topicPartition, currentSequenceNumber); + nextSequence.put(topicPartition, currentSequenceNumber); + } + + synchronized void addInFlightBatch(ProducerBatch batch) { + if (!batch.hasSequence()) + throw new IllegalStateException("Can't track batch for partition " + batch.topicPartition + " when sequence is not set."); + if (!inflightBatchesBySequence.containsKey(batch.topicPartition)) { + inflightBatchesBySequence.put(batch.topicPartition, new PriorityQueue<>(5, new Comparator() { + @Override + public int compare(ProducerBatch o1, ProducerBatch o2) { + return o1.baseSequence() - o2.baseSequence(); + } + })); + } + inflightBatchesBySequence.get(batch.topicPartition).offer(batch); + } + + + synchronized ProducerBatch nextBatchBySequence(TopicPartition topicPartition) { + PriorityQueue queue = inflightBatchesBySequence.get(topicPartition); + if (queue == null) + return null; + return queue.peek(); + } + + synchronized void removeInFlightBatch(ProducerBatch batch) { + PriorityQueue queue = inflightBatchesBySequence.get(batch.topicPartition); + if (queue == null) + return; + queue.remove(batch); + } + + synchronized void maybeUpdateLastAckedSequence(TopicPartition topicPartition, int sequence) { + if (sequence > lastAckedSequence(topicPartition)) + lastAckedSequence.put(topicPartition, sequence); + } + + synchronized int lastAckedSequence(TopicPartition topicPartition) { + Integer currentLastAckedSequence = lastAckedSequence.get(topicPartition); + if (currentLastAckedSequence == null) + return -1; + return currentLastAckedSequence; + } + + // If a batch is failed fatally, the sequence numbers for future batches bound for the partition must be adjusted + // so that they don't fail with the OutOfOrderSequenceException. + // + // This method must only be called when we know that the batch is question has been unequivocally failed by the broker, + // ie. it has received a confirmed fatal status code like 'Message Too Large' or something similar. + synchronized void adjustSequencesDueToFailedBatch(ProducerBatch batch) { + if (!this.nextSequence.containsKey(batch.topicPartition)) + // Sequence numbers are not being tracked for this partition. This could happen if the producer id was just + // reset due to a previous OutOfOrderSequenceException. + return; + log.debug("producerId: {}, send to partition {} failed fatally. Reducing future sequence numbers by {}", + batch.producerId(), batch.topicPartition, batch.recordCount); + int currentSequence = sequenceNumber(batch.topicPartition); + currentSequence -= batch.recordCount; + if (currentSequence < 0) + throw new IllegalStateException("Sequence number for partition " + batch.topicPartition + " is going to become negative : " + currentSequence); + + setNextSequence(batch.topicPartition, currentSequence); + + for (ProducerBatch inFlightBatch : inflightBatchesBySequence.get(batch.topicPartition)) { + if (inFlightBatch.baseSequence() < batch.baseSequence()) + continue; + int newSequence = inFlightBatch.baseSequence() - batch.recordCount; + if (newSequence < 0) + throw new IllegalStateException("Sequence number for batch with sequence " + inFlightBatch.baseSequence() + + " for partition " + batch.topicPartition + " is going to become negative :" + newSequence); + + log.info("Resetting sequence number of batch with current sequence {} for partition {} to {}", inFlightBatch.baseSequence(), batch.topicPartition, newSequence); + inFlightBatch.resetProducerState(new ProducerIdAndEpoch(inFlightBatch.producerId(), inFlightBatch.producerEpoch()), newSequence, inFlightBatch.isTransactional()); + } + } + + synchronized boolean hasInflightBatches(TopicPartition topicPartition) { + return inflightBatchesBySequence.containsKey(topicPartition) && !inflightBatchesBySequence.get(topicPartition).isEmpty(); + } + + synchronized boolean hasUnresolvedSequences() { + return !partitionsWithUnresolvedSequences.isEmpty(); + } + + synchronized boolean hasUnresolvedSequence(TopicPartition topicPartition) { + return partitionsWithUnresolvedSequences.contains(topicPartition); + } + + synchronized void markSequenceUnresolved(TopicPartition topicPartition) { + log.debug("Marking partition {} unresolved", topicPartition); + partitionsWithUnresolvedSequences.add(topicPartition); + } + + // Checks if there are any partitions with unresolved partitions which may now be resolved. Returns true if + // the producer id needs a reset, false otherwise. + synchronized boolean shouldResetProducerStateAfterResolvingSequences() { + if (isTransactional()) + // We should not reset producer state if we are transactional. We will transition to a fatal error instead. + return false; + for (TopicPartition topicPartition : partitionsWithUnresolvedSequences) { + if (!hasInflightBatches(topicPartition)) { + // The partition has been fully drained. At this point, the last ack'd sequence should be once less than + // next sequence destined for the partition. If so, the partition is fully resolved. If not, we should + // reset the sequence number if necessary. + if (isNextSequence(topicPartition, sequenceNumber(topicPartition))) { + // This would happen when a batch was expired, but subsequent batches succeeded. + partitionsWithUnresolvedSequences.remove(topicPartition); + } else { + // We would enter this branch if all in flight batches were ultimately expired in the producer. + log.info("No inflight batches remaining for {}, last ack'd sequence for partition is {}, next sequence is {}. " + + "Going to reset producer state.", topicPartition, lastAckedSequence(topicPartition), sequenceNumber(topicPartition)); + return true; + } + } + } + return false; + } + + synchronized boolean isNextSequence(TopicPartition topicPartition, int sequence) { + return sequence - lastAckedSequence(topicPartition) == 1; + } + + private synchronized void setNextSequence(TopicPartition topicPartition, int sequence) { + if (!nextSequence.containsKey(topicPartition) && sequence != 0) + throw new IllegalStateException("Trying to set the sequence number for " + topicPartition + " to " + sequence + + ", but the sequence number was never set for this partition."); + nextSequence.put(topicPartition, sequence); } synchronized TxnRequestHandler nextRequestHandler(boolean hasIncompleteBatches) { @@ -441,15 +594,15 @@ void lookupCoordinator(TxnRequestHandler request) { lookupCoordinator(request.coordinatorType(), request.coordinatorKey()); } - void setInFlightRequestCorrelationId(int correlationId) { + void setInFlightTransactionalRequestCorrelationId(int correlationId) { inFlightRequestCorrelationId = correlationId; } - void clearInFlightRequestCorrelationId() { + void clearInFlightTransactionalRequestCorrelationId() { inFlightRequestCorrelationId = NO_INFLIGHT_REQUEST_CORRELATION_ID; } - boolean hasInFlightRequest() { + boolean hasInFlightTransactionalRequest() { return inFlightRequestCorrelationId != NO_INFLIGHT_REQUEST_CORRELATION_ID; } @@ -479,6 +632,11 @@ synchronized boolean hasOngoingTransaction() { return currentState == State.IN_TRANSACTION || isCompleting() || hasAbortableError(); } + synchronized boolean canRetryOutOfOrderSequenceException(ProducerBatch batch) { + return hasProducerId(batch.producerId()) && !hasUnresolvedSequence(batch.topicPartition) && + (batch.sequenceHasBeenReset() || !isNextSequence(batch.topicPartition, batch.baseSequence())); + } + // visible for testing synchronized boolean isReady() { return isTransactional() && currentState == State.READY; @@ -629,7 +787,7 @@ public void onComplete(ClientResponse response) { if (response.requestHeader().correlationId() != inFlightRequestCorrelationId) { fatalError(new RuntimeException("Detected more than one in-flight transactional request.")); } else { - clearInFlightRequestCorrelationId(); + clearInFlightTransactionalRequestCorrelationId(); if (response.wasDisconnected()) { log.debug("Disconnected from {}. Will retry.", response.destination()); if (this.needsCoordinator()) diff --git a/clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceNumberException.java b/clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceException.java similarity index 86% rename from clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceNumberException.java rename to clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceException.java index 469ba98f445e9..11f81af5bb868 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceNumberException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/DuplicateSequenceException.java @@ -16,9 +16,9 @@ */ package org.apache.kafka.common.errors; -public class DuplicateSequenceNumberException extends RetriableException { +public class DuplicateSequenceException extends ApiException { - public DuplicateSequenceNumberException(String message) { + public DuplicateSequenceException(String message) { super(message); } } diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java index 9decef277626b..bbc348643e0cc 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Errors.java @@ -25,7 +25,7 @@ import org.apache.kafka.common.errors.CoordinatorNotAvailableException; import org.apache.kafka.common.errors.CorruptRecordException; import org.apache.kafka.common.errors.LogDirNotFoundException; -import org.apache.kafka.common.errors.DuplicateSequenceNumberException; +import org.apache.kafka.common.errors.DuplicateSequenceException; import org.apache.kafka.common.errors.GroupAuthorizationException; import org.apache.kafka.common.errors.IllegalGenerationException; import org.apache.kafka.common.errors.IllegalSaslStateException; @@ -432,7 +432,7 @@ public ApiException build(String message) { new ApiExceptionBuilder() { @Override public ApiException build(String message) { - return new DuplicateSequenceNumberException(message); + return new DuplicateSequenceException(message); } }), INVALID_PRODUCER_EPOCH(47, "Producer attempted an operation with an old epoch. Either there is a newer producer " + diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java index 19d25d7b52dc8..fc83134ebe4e8 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java @@ -279,6 +279,17 @@ public void abort() { aborted = true; } + public void reopenAndRewriteProducerState(long producerId, short producerEpoch, int baseSequence, boolean isTransactional) { + if (aborted) + throw new IllegalStateException("Should not reopen a batch which is already aborted."); + builtRecords = null; + this.producerId = producerId; + this.producerEpoch = producerEpoch; + this.baseSequence = baseSequence; + this.isTransactional = isTransactional; + } + + public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); @@ -766,4 +777,7 @@ public short producerEpoch() { return this.producerEpoch; } + public int baseSequence() { + return this.baseSequence; + } } diff --git a/clients/src/test/java/org/apache/kafka/clients/MockClient.java b/clients/src/test/java/org/apache/kafka/clients/MockClient.java index 9960ccef953d1..71e32ff759a38 100644 --- a/clients/src/test/java/org/apache/kafka/clients/MockClient.java +++ b/clients/src/test/java/org/apache/kafka/clients/MockClient.java @@ -224,6 +224,16 @@ public void respond(RequestMatcher matcher, AbstractResponse response) { respond(response); } + // Utility method to enable out of order responses + public void respondToRequest(ClientRequest clientRequest, AbstractResponse response) { + AbstractRequest request = clientRequest.requestBuilder().build(); + requests.remove(clientRequest); + short version = clientRequest.requestBuilder().desiredOrLatestVersion(); + responses.add(new ClientResponse(clientRequest.makeHeader(version), clientRequest.callback(), clientRequest.destination(), + clientRequest.createdTimeMs(), time.milliseconds(), false, null, response)); + } + + public void respond(AbstractResponse response, boolean disconnected) { ClientRequest request = requests.remove(); short version = request.requestBuilder().desiredOrLatestVersion(); diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java index 6f98e522621e2..26e3e6c49e8cc 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java @@ -29,6 +29,8 @@ import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.ClusterAuthorizationException; +import org.apache.kafka.common.errors.OutOfOrderSequenceException; +import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.common.errors.TimeoutException; import org.apache.kafka.common.errors.UnsupportedForMessageFormatException; import org.apache.kafka.common.errors.UnsupportedVersionException; @@ -66,6 +68,7 @@ import java.nio.ByteBuffer; import java.util.Collections; +import java.util.Deque; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; @@ -467,6 +470,657 @@ public void testClusterAuthorizationExceptionInInitProducerIdRequest() throws Ex assertSendFailure(ClusterAuthorizationException.class); } + + @Test + public void testIdempotenceWithMultipleInflights() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0L); + + sender.run(time.milliseconds()); // receive response 0 + + assertEquals(1, client.inFlightRequestCount()); + assertEquals(0, transactionManager.lastAckedSequence(tp0)); + assertTrue(request1.isDone()); + assertEquals(0, request1.get().offset()); + assertFalse(request2.isDone()); + + sendIdempotentProducerResponse(1, tp0, Errors.NONE, 1L); + sender.run(time.milliseconds()); // receive response 1 + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertFalse(client.hasInFlightRequests()); + assertTrue(request2.isDone()); + assertEquals(1, request2.get().offset()); + } + + + @Test + public void testIdempotenceWithMultipleInflightsFirstFails() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + sendIdempotentProducerResponse(0, tp0, Errors.LEADER_NOT_AVAILABLE, -1L); + + sender.run(time.milliseconds()); // receive response 0 + + assertEquals(1, client.inFlightRequestCount()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + sendIdempotentProducerResponse(1, tp0, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, -1L); + + sender.run(time.milliseconds()); // re send request 0, receive response 1 + + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertEquals(1, client.inFlightRequestCount()); + + sender.run(time.milliseconds()); // Do nothing, we are reduced to one in flight request during retries. + + assertEquals(1, client.inFlightRequestCount()); + + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0L); + sender.run(time.milliseconds()); // receive response 0 + assertEquals(0, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + + assertFalse(request2.isDone()); + assertTrue(request1.isDone()); + assertEquals(0, request1.get().offset()); + + sender.run(time.milliseconds()); // send request 1 + assertEquals(1, client.inFlightRequestCount()); + sendIdempotentProducerResponse(1, tp0, Errors.NONE, 1L); + sender.run(time.milliseconds()); // receive response 1 + + assertTrue(request2.isDone()); + assertEquals(1, request2.get().offset()); + assertFalse(client.hasInFlightRequests()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + } + + @Test + public void testIdempotenceWithMultipleInflightsWhereFirstFailsFatallyAndSequenceOfFutureBatchesIsAdjusted() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + sendIdempotentProducerResponse(0, tp0, Errors.MESSAGE_TOO_LARGE, -1L); + + sender.run(time.milliseconds()); // receive response 0, should adjust sequences of future batches. + + assertTrue(request1.isDone()); + try { + request1.get(); + fail("Should have raised an error"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof RecordTooLargeException); + } + + assertEquals(1, client.inFlightRequestCount()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + sendIdempotentProducerResponse(1, tp0, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, -1L); + + sender.run(time.milliseconds()); // receive response 1 + + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + + sender.run(time.milliseconds()); // resend request 1 + + assertEquals(1, client.inFlightRequestCount()); + + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0L); + sender.run(time.milliseconds()); // receive response 1 + assertEquals(0, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + + assertTrue(request1.isDone()); + assertEquals(0, request2.get().offset()); + } + + @Test + public void testMustNotRetryOutOfOrderSequenceForNextBatch() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest with multiple messages. + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT); + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + + // make sure the next sequence number accounts for multi-message batches. + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0); + + sender.run(time.milliseconds()); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(3, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertTrue(request1.isDone()); + assertEquals(0, request1.get().offset()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + // This OutOfOrderSequence is fatal since it is returned for the batch succeeding the last acknowledged batch. + sendIdempotentProducerResponse(2, tp0, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, -1L); + + sender.run(time.milliseconds()); + assertTrue(request2.isDone()); + + try { + request2.get(); + fail("Expected an OutOfOrderSequenceException"); + } catch (ExecutionException e) { + assert e.getCause() instanceof OutOfOrderSequenceException; + } + } + + @Test + public void testCorrectHandlingOfOutOfOrderResponses() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + ClientRequest firstClientRequest = client.requests().peek(); + ClientRequest secondClientRequest = (ClientRequest) client.requests().toArray()[1]; + + client.respondToRequest(secondClientRequest, produceResponse(tp0, -1, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, -1)); + + sender.run(time.milliseconds()); // receive response 1 + Deque queuedBatches = accumulator.batches().get(tp0); + + // Make sure that we are queueing the second batch first. + assertEquals(1, queuedBatches.size()); + assertEquals(1, queuedBatches.peekFirst().baseSequence()); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + client.respondToRequest(firstClientRequest, produceResponse(tp0, -1, Errors.NOT_LEADER_FOR_PARTITION, -1)); + + sender.run(time.milliseconds()); // receive response 0 + + // Make sure we requeued both batches in the correct order. + assertEquals(2, queuedBatches.size()); + assertEquals(0, queuedBatches.peekFirst().baseSequence()); + assertEquals(1, queuedBatches.peekLast().baseSequence()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + + sender.run(time.milliseconds()); // send request 0 + assertEquals(1, client.inFlightRequestCount()); + sender.run(time.milliseconds()); // don't do anything, only one inflight allowed once we are retrying. + + assertEquals(1, client.inFlightRequestCount()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Make sure that the requests are sent in order, even though the previous responses were not in order. + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0L); + sender.run(time.milliseconds()); // receive response 0 + assertEquals(0, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + assertTrue(request1.isDone()); + assertEquals(0, request1.get().offset()); + + sender.run(time.milliseconds()); // send request 1 + assertEquals(1, client.inFlightRequestCount()); + sendIdempotentProducerResponse(1, tp0, Errors.NONE, 1L); + sender.run(time.milliseconds()); // receive response 1 + + assertFalse(client.hasInFlightRequests()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertTrue(request2.isDone()); + assertEquals(1, request2.get().offset()); + } + + @Test + public void testCorrectHandlingOfOutOfOrderResponsesWhenSecondSucceeds() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + ClientRequest firstClientRequest = client.requests().peek(); + ClientRequest secondClientRequest = (ClientRequest) client.requests().toArray()[1]; + + client.respondToRequest(secondClientRequest, produceResponse(tp0, 1, Errors.NONE, 1)); + + sender.run(time.milliseconds()); // receive response 1 + assertTrue(request2.isDone()); + assertEquals(1, request2.get().offset()); + assertFalse(request1.isDone()); + Deque queuedBatches = accumulator.batches().get(tp0); + + assertEquals(0, queuedBatches.size()); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + + client.respondToRequest(firstClientRequest, produceResponse(tp0, -1, Errors.REQUEST_TIMED_OUT, -1)); + + sender.run(time.milliseconds()); // receive response 0 + + // Make sure we requeued both batches in the correct order. + assertEquals(1, queuedBatches.size()); + assertEquals(0, queuedBatches.peekFirst().baseSequence()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + + sender.run(time.milliseconds()); // resend request 0 + assertEquals(1, client.inFlightRequestCount()); + + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + + // Make sure we handle the out of order successful responses correctly. + sendIdempotentProducerResponse(0, tp0, Errors.NONE, 0L); + sender.run(time.milliseconds()); // receive response 0 + assertEquals(0, queuedBatches.size()); + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertEquals(0, client.inFlightRequestCount()); + + assertFalse(client.hasInFlightRequests()); + assertTrue(request1.isDone()); + assertEquals(0, request1.get().offset()); + } + + @Test + public void testExpiryOfUnsentBatchesShouldNotCauseUnresolvedSequences() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, 0L, "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + Node node = this.cluster.nodes().get(0); + time.sleep(10000L); + client.disconnect(node.idString()); + client.blackout(node, 10); + + sender.run(time.milliseconds()); + + assertTrue(request1.isDone()); + try { + request1.get(); + fail("Should have raised timeout exception"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + assertFalse(transactionManager.hasUnresolvedSequence(tp0)); + } + + @Test + public void testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatchesSucceed() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + + assertEquals(2, client.inFlightRequestCount()); + + sendIdempotentProducerResponse(0, tp0, Errors.REQUEST_TIMED_OUT, -1); + sender.run(time.milliseconds()); // receive first response + + Node node = this.cluster.nodes().get(0); + time.sleep(10000L); + client.disconnect(node.idString()); + client.blackout(node, 10); + + sender.run(time.milliseconds()); // now expire the first batch. + assertTrue(request1.isDone()); + try { + request1.get(); + fail("Should have raised timeout exception"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + assertTrue(transactionManager.hasUnresolvedSequence(tp0)); + // let's enqueue another batch, which should not be dequeued until the unresolved state is clear. + Future request3 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + + time.sleep(20); + + assertFalse(request2.isDone()); + + sender.run(time.milliseconds()); // send second request + sendIdempotentProducerResponse(1, tp0, Errors.NONE, 1); + sender.run(time.milliseconds()); // receive second response, the third request shouldn't be sent since we are in an unresolved state. + assertTrue(request2.isDone()); + assertEquals(1, request2.get().offset()); + Deque batches = accumulator.batches().get(tp0); + + assertEquals(1, batches.size()); + assertFalse(batches.peekFirst().hasSequence()); + assertFalse(client.hasInFlightRequests()); + assertEquals(2L, transactionManager.sequenceNumber(tp0).longValue()); + assertTrue(transactionManager.hasUnresolvedSequence(tp0)); + + sender.run(time.milliseconds()); // clear the unresolved state, send the pending request. + assertFalse(transactionManager.hasUnresolvedSequence(tp0)); + assertTrue(transactionManager.hasProducerId()); + assertEquals(0, batches.size()); + assertEquals(1, client.inFlightRequestCount()); + assertFalse(request3.isDone()); + } + + @Test + public void testExpiryOfFirstBatchShouldCauseResetIfFutureBatchesFail() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + + assertEquals(2, client.inFlightRequestCount()); + + sendIdempotentProducerResponse(0, tp0, Errors.NOT_LEADER_FOR_PARTITION, -1); + sender.run(time.milliseconds()); // receive first response + + Node node = this.cluster.nodes().get(0); + time.sleep(10000L); + client.disconnect(node.idString()); + client.blackout(node, 10); + + sender.run(time.milliseconds()); // now expire the first batch. + assertTrue(request1.isDone()); + try { + request1.get(); + fail("Should have raised timeout exception"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + assertTrue(transactionManager.hasUnresolvedSequence(tp0)); + // let's enqueue another batch, which should not be dequeued until the unresolved state is clear. + Future request3 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + + time.sleep(20); + + assertFalse(request2.isDone()); + + sender.run(time.milliseconds()); // send second request + sendIdempotentProducerResponse(1, tp0, Errors.OUT_OF_ORDER_SEQUENCE_NUMBER, 1); + sender.run(time.milliseconds()); // receive second response, the third request shouldn't be sent since we are in an unresolved state. + assertTrue(request2.isDone()); + + try { + request2.get(); + fail("should have failed with an exception"); + } catch (Exception e) { + assertTrue(e.getCause() instanceof OutOfOrderSequenceException); + } + + Deque batches = accumulator.batches().get(tp0); + + // The second request should not be requeued. + assertEquals(1, batches.size()); + assertFalse(batches.peekFirst().hasSequence()); + assertFalse(client.hasInFlightRequests()); + + // The producer state should be reset. + assertFalse(transactionManager.hasProducerId()); + assertFalse(transactionManager.hasUnresolvedSequence(tp0)); + } + + @Test + public void testExpiryOfAllSentBatchesShouldCauseUnresolvedSequences() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, 0L, "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + sendIdempotentProducerResponse(0, tp0, Errors.NOT_LEADER_FOR_PARTITION, -1); + sender.run(time.milliseconds()); // receive response + + assertEquals(1L, transactionManager.sequenceNumber(tp0).longValue()); + + Node node = this.cluster.nodes().get(0); + time.sleep(10000L); + client.disconnect(node.idString()); + client.blackout(node, 10); + + sender.run(time.milliseconds()); // now expire the batch. + + assertTrue(request1.isDone()); + try { + request1.get(); + fail("Should have raised timeout exception"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + assertTrue(transactionManager.hasUnresolvedSequence(tp0)); + assertFalse(client.hasInFlightRequests()); + Deque batches = accumulator.batches().get(tp0); + assertEquals(0, batches.size()); + assertTrue(transactionManager.hasProducerId(producerId)); + // We should now clear the old producerId and get a new one in a single run loop. + prepareAndReceiveInitProducerId(producerId + 1, Errors.NONE); + assertTrue(transactionManager.hasProducerId(producerId + 1)); + } + + @Test + public void testCorrectHandlingOfDuplicateSequenceError() throws Exception { + final long producerId = 343434L; + TransactionManager transactionManager = new TransactionManager(); + setupWithTransactionState(transactionManager); + prepareAndReceiveInitProducerId(producerId, Errors.NONE); + assertTrue(transactionManager.hasProducerId()); + + assertEquals(0, transactionManager.sequenceNumber(tp0).longValue()); + + // Send first ProduceRequest + Future request1 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + String nodeId = client.requests().peek().destination(); + Node node = new Node(Integer.valueOf(nodeId), "localhost", 0); + assertEquals(1, client.inFlightRequestCount()); + assertEquals(1, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + + // Send second ProduceRequest + Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); + assertEquals(2, client.inFlightRequestCount()); + assertEquals(2, transactionManager.sequenceNumber(tp0).longValue()); + assertEquals(-1, transactionManager.lastAckedSequence(tp0)); + assertFalse(request1.isDone()); + assertFalse(request2.isDone()); + assertTrue(client.isReady(node, time.milliseconds())); + + ClientRequest firstClientRequest = client.requests().peek(); + ClientRequest secondClientRequest = (ClientRequest) client.requests().toArray()[1]; + + client.respondToRequest(secondClientRequest, produceResponse(tp0, 1, Errors.NONE, -1)); + + sender.run(time.milliseconds()); // receive response 1 + + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + + client.respondToRequest(firstClientRequest, produceResponse(tp0, -1, Errors.DUPLICATE_SEQUENCE_NUMBER, -1)); + + sender.run(time.milliseconds()); // receive response 0 + + // Make sure that the last ack'd sequence doesn't change. + assertEquals(1, transactionManager.lastAckedSequence(tp0)); + assertFalse(client.hasInFlightRequests()); + } + + void sendIdempotentProducerResponse(final int expectedSequence, TopicPartition tp, Errors responseError, long responseOffset) { + client.respond(new MockClient.RequestMatcher() { + @Override + public boolean matches(AbstractRequest body) { + ProduceRequest produceRequest = (ProduceRequest) body; + assertTrue(produceRequest.isIdempotent()); + + MemoryRecords records = produceRequest.partitionRecordsOrFail().get(tp0); + Iterator batchIterator = records.batches().iterator(); + RecordBatch firstBatch = batchIterator.next(); + assertFalse(batchIterator.hasNext()); + assertEquals(expectedSequence, firstBatch.baseSequence()); + + return true; + } + }, produceResponse(tp, responseOffset, responseError, 0)); + } + @Test public void testClusterAuthorizationExceptionInProduceRequest() throws Exception { final long producerId = 343434L; @@ -604,7 +1258,8 @@ public boolean matches(AbstractRequest body) { sender.run(time.milliseconds()); // receive response assertTrue(responseFuture.isDone()); - assertEquals((long) transactionManager.sequenceNumber(tp0), 1L); + assertEquals(0L, (long) transactionManager.lastAckedSequence(tp0)); + assertEquals(1L, (long) transactionManager.sequenceNumber(tp0)); } @Test @@ -632,6 +1287,7 @@ public void testAbortRetryWhenProducerIdChanges() throws InterruptedException { assertEquals(0, client.inFlightRequestCount()); assertFalse("Client ready status should be false", client.isReady(node, 0L)); + transactionManager.resetProducerId(); transactionManager.setProducerIdAndEpoch(new ProducerIdAndEpoch(producerId + 1, (short) 0)); sender.run(time.milliseconds()); // receive error sender.run(time.milliseconds()); // reconnect @@ -642,7 +1298,7 @@ public void testAbortRetryWhenProducerIdChanges() throws InterruptedException { assertTrue("Expected non-zero value for record send errors", recordErrors.value() > 0); assertTrue(responseFuture.isDone()); - assertEquals((long) transactionManager.sequenceNumber(tp0), 0L); + assertEquals(0, (long) transactionManager.sequenceNumber(tp0)); } @Test @@ -720,7 +1376,8 @@ private void testSplitBatchAndSend(TransactionManager txnManager, accumulator.append(tp, 0L, "key2".getBytes(), new byte[batchSize / 2], null, null, MAX_BLOCK_TIMEOUT).future; sender.run(time.milliseconds()); // connect sender.run(time.milliseconds()); // send produce request - assertEquals("The sequence number should be 0", 0, txnManager.sequenceNumber(tp).longValue()); + + assertEquals("The next sequence should be 2", 2, txnManager.sequenceNumber(tp).longValue()); String id = client.requests().peek().destination(); assertEquals(ApiKeys.PRODUCE, client.requests().peek().requestBuilder().apiKey()); Node node = new Node(Integer.valueOf(id), "localhost", 0); @@ -731,11 +1388,12 @@ private void testSplitBatchAndSend(TransactionManager txnManager, responseMap.put(tp, new ProduceResponse.PartitionResponse(Errors.MESSAGE_TOO_LARGE)); client.respond(new ProduceResponse(responseMap)); sender.run(time.milliseconds()); // split and reenqueue + assertEquals("The next sequence should be 2", 2, txnManager.sequenceNumber(tp).longValue()); // The compression ratio should have been improved once. assertEquals(CompressionType.GZIP.rate - CompressionRatioEstimator.COMPRESSION_RATIO_IMPROVING_STEP, CompressionRatioEstimator.estimation(topic, CompressionType.GZIP), 0.01); - sender.run(time.milliseconds()); // send produce request - assertEquals("The sequence number should be 0", 0, txnManager.sequenceNumber(tp).longValue()); + sender.run(time.milliseconds()); // send the first produce request + assertEquals("The next sequence number should be 2", 2, txnManager.sequenceNumber(tp).longValue()); assertFalse("The future shouldn't have been done.", f1.isDone()); assertFalse("The future shouldn't have been done.", f2.isDone()); id = client.requests().peek().destination(); @@ -750,10 +1408,11 @@ private void testSplitBatchAndSend(TransactionManager txnManager, sender.run(time.milliseconds()); // receive assertTrue("The future should have been done.", f1.isDone()); - assertEquals("The sequence number should be 1", 1, txnManager.sequenceNumber(tp).longValue()); + assertEquals("The next sequence number should still be 2", 2, txnManager.sequenceNumber(tp).longValue()); + assertEquals("The last ack'd sequence number should be 0", 0, txnManager.lastAckedSequence(tp)); assertFalse("The future shouldn't have been done.", f2.isDone()); assertEquals("Offset of the first message should be 0", 0L, f1.get().offset()); - sender.run(time.milliseconds()); // send produce request + sender.run(time.milliseconds()); // send the seconcd produce request id = client.requests().peek().destination(); assertEquals(ApiKeys.PRODUCE, client.requests().peek().requestBuilder().apiKey()); node = new Node(Integer.valueOf(id), "localhost", 0); @@ -766,7 +1425,8 @@ private void testSplitBatchAndSend(TransactionManager txnManager, sender.run(time.milliseconds()); // receive assertTrue("The future should have been done.", f2.isDone()); - assertEquals("The sequence number should be 2", 2, txnManager.sequenceNumber(tp).longValue()); + assertEquals("The next sequence number should be 2", 2, txnManager.sequenceNumber(tp).longValue()); + assertEquals("The last ack'd sequence number should be 1", 1, txnManager.lastAckedSequence(tp)); assertEquals("Offset of the first message should be 1", 1L, f2.get().offset()); assertTrue("There should be no batch in the accumulator", accumulator.batches().get(tp).isEmpty()); @@ -828,9 +1488,11 @@ private void setupWithTransactionState(TransactionManager transactionManager) { this.metrics = new Metrics(metricConfig, time); this.accumulator = new RecordAccumulator(logContext, batchSize, 1024 * 1024, CompressionType.NONE, 0L, 0L, metrics, time, apiVersions, transactionManager); + this.senderMetricsRegistry = new SenderMetricsRegistry(metricTags.keySet()); - this.sender = new Sender(logContext, this.client, this.metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, - MAX_RETRIES, this.metrics, this.senderMetricsRegistry, this.time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); + + this.sender = new Sender(logContext, this.client, this.metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL, + Integer.MAX_VALUE, this.metrics, this.senderMetricsRegistry, this.time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); this.metadata.update(this.cluster, Collections.emptySet(), time.milliseconds()); } diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java index 53bba1c13fdad..28f9c820c4a15 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java @@ -1171,6 +1171,7 @@ public void testRecoveryFromAbortableErrorProduceRequestInRetry() throws Excepti assertFutureFailed(unauthorizedTopicProduceFuture); assertTrue(authorizedTopicProduceFuture.isDone()); assertNotNull(authorizedTopicProduceFuture.get()); + assertTrue(authorizedTopicProduceFuture.isDone()); prepareEndTxnResponse(Errors.NONE, TransactionResult.ABORT, pid, epoch); transactionManager.beginAbort(); @@ -1481,14 +1482,14 @@ public void testCommitTransactionWithUnsentProduceRequest() throws Exception { sender.run(time.milliseconds()); assertFalse(accumulator.hasUndrained()); assertTrue(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertFalse(responseFuture.isDone()); // until the produce future returns, we will not send EndTxn sender.run(time.milliseconds()); assertFalse(accumulator.hasUndrained()); assertTrue(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertFalse(responseFuture.isDone()); // now the produce response returns @@ -1497,14 +1498,14 @@ public void testCommitTransactionWithUnsentProduceRequest() throws Exception { assertTrue(responseFuture.isDone()); assertFalse(accumulator.hasUndrained()); assertFalse(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); // now we send EndTxn sender.run(time.milliseconds()); - assertTrue(transactionManager.hasInFlightRequest()); + assertTrue(transactionManager.hasInFlightTransactionalRequest()); sendEndTxnResponse(Errors.NONE, TransactionResult.COMMIT, pid, epoch); sender.run(time.milliseconds()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertTrue(transactionManager.isReady()); } @@ -1529,21 +1530,21 @@ public void testCommitTransactionWithInFlightProduceRequest() throws Exception { sender.run(time.milliseconds()); assertFalse(accumulator.hasUndrained()); assertTrue(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); // now we begin the commit with the produce request still pending transactionManager.beginCommit(); sender.run(time.milliseconds()); assertFalse(accumulator.hasUndrained()); assertTrue(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertFalse(responseFuture.isDone()); // until the produce future returns, we will not send EndTxn sender.run(time.milliseconds()); assertFalse(accumulator.hasUndrained()); assertTrue(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertFalse(responseFuture.isDone()); // now the produce response returns @@ -1552,14 +1553,14 @@ public void testCommitTransactionWithInFlightProduceRequest() throws Exception { assertTrue(responseFuture.isDone()); assertFalse(accumulator.hasUndrained()); assertFalse(accumulator.hasIncomplete()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); // now we send EndTxn sender.run(time.milliseconds()); - assertTrue(transactionManager.hasInFlightRequest()); + assertTrue(transactionManager.hasInFlightTransactionalRequest()); sendEndTxnResponse(Errors.NONE, TransactionResult.COMMIT, pid, epoch); sender.run(time.milliseconds()); - assertFalse(transactionManager.hasInFlightRequest()); + assertFalse(transactionManager.hasInFlightTransactionalRequest()); assertTrue(transactionManager.isReady()); } @@ -1960,17 +1961,17 @@ public void resendFailedProduceRequestAfterAbortableError() throws Exception { prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, pid); prepareProduceResponse(Errors.NOT_LEADER_FOR_PARTITION, pid, epoch); - sender.run(time.milliseconds()); // AddPartitions + sender.run(time.milliseconds()); // Add partitions sender.run(time.milliseconds()); // Produce assertFalse(responseFuture.isDone()); transactionManager.transitionToAbortableError(new KafkaException()); prepareProduceResponse(Errors.NONE, pid, epoch); - sender.run(time.milliseconds()); + assertTrue(responseFuture.isDone()); - assertNotNull(responseFuture.get()); + assertNotNull(responseFuture.get()); // should throw the exception which caused the transaction to be aborted. } @Test @@ -2151,6 +2152,65 @@ public void testDropCommitOnBatchExpiry() throws InterruptedException, Execution assertFalse(transactionManager.transactionContainsPartition(tp0)); } + @Test + public void testTransitionToFatalErrorWhenRetriedBatchIsExpired() throws InterruptedException, ExecutionException { + final long pid = 13131L; + final short epoch = 1; + + doInitTransactions(pid, epoch); + + transactionManager.beginTransaction(); + transactionManager.maybeAddPartitionToTransaction(tp0); + + Future responseFuture = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), + "value".getBytes(), Record.EMPTY_HEADERS, null, MAX_BLOCK_TIMEOUT).future; + + assertFalse(responseFuture.isDone()); + + prepareAddPartitionsToTxnResponse(Errors.NONE, tp0, epoch, pid); + + assertFalse(transactionManager.transactionContainsPartition(tp0)); + assertFalse(transactionManager.isSendToPartitionAllowed(tp0)); + sender.run(time.milliseconds()); // send addPartitions. + // Check that only addPartitions was sent. + assertTrue(transactionManager.transactionContainsPartition(tp0)); + assertTrue(transactionManager.isSendToPartitionAllowed(tp0)); + + prepareProduceResponse(Errors.NOT_LEADER_FOR_PARTITION, pid, epoch); + sender.run(time.milliseconds()); // send the produce request. + + assertFalse(responseFuture.isDone()); + + TransactionalRequestResult commitResult = transactionManager.beginCommit(); + + // Sleep 10 seconds to make sure that the batches in the queue would be expired if they can't be drained. + time.sleep(10000); + // Disconnect the target node for the pending produce request. This will ensure that sender will try to + // expire the batch. + Node clusterNode = this.cluster.nodes().get(0); + client.disconnect(clusterNode.idString()); + client.blackout(clusterNode, 100); + + sender.run(time.milliseconds()); // We should try to flush the produce, but expire it instead without sending anything. + assertTrue(responseFuture.isDone()); + + try { + // make sure the produce was expired. + responseFuture.get(); + fail("Expected to get a TimeoutException since the queued ProducerBatch should have been expired"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + sender.run(time.milliseconds()); // Transition to fatal error since we have unresolved batches. + sender.run(time.milliseconds()); // Fail the queued transactional requests + + assertTrue(commitResult.isCompleted()); + assertFalse(commitResult.isSuccessful()); // the commit should have been dropped. + + assertTrue(transactionManager.hasFatalError()); + assertFalse(transactionManager.hasOngoingTransaction()); + } + private void verifyAddPartitionsFailsWithPartitionLevelError(final Errors error) throws InterruptedException { final long pid = 1L; final short epoch = 1; diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala index d3de24db643c4..d98f443e2629d 100644 --- a/core/src/main/scala/kafka/log/Log.scala +++ b/core/src/main/scala/kafka/log/Log.scala @@ -514,8 +514,10 @@ class Log(@volatile var dir: File, completedTxns.foreach(producerStateManager.completeTxn) } - private[log] def activeProducers: Map[Long, ProducerIdEntry] = lock synchronized { - producerStateManager.activeProducers + private[log] def activeProducersWithLastSequence: Map[Long, Int] = lock synchronized { + producerStateManager.activeProducers.map { case (producerId, producerIdEntry) => + (producerId, producerIdEntry.lastSeq) + } } /** @@ -765,17 +767,18 @@ class Log(@volatile var dir: File, } private def analyzeAndValidateProducerState(records: MemoryRecords, isFromClient: Boolean): - (mutable.Map[Long, ProducerAppendInfo], List[CompletedTxn], Option[ProducerIdEntry]) = { + (mutable.Map[Long, ProducerAppendInfo], List[CompletedTxn], Option[BatchMetadata]) = { val updatedProducers = mutable.Map.empty[Long, ProducerAppendInfo] val completedTxns = ListBuffer.empty[CompletedTxn] for (batch <- records.batches.asScala if batch.hasProducerId) { val maybeLastEntry = producerStateManager.lastEntry(batch.producerId) - // if this is a client produce request, there will be only one batch. If that batch matches - // the last appended entry for that producer, then this request is a duplicate and we return - // the last appended entry to the client. - if (isFromClient && maybeLastEntry.exists(_.isDuplicate(batch))) - return (updatedProducers, completedTxns.toList, maybeLastEntry) + // if this is a client produce request, there will be upto 5 batches which could have been duplicated. + // If we find a duplicate, we return the metadata of the appended batch to the client. + if (isFromClient) + maybeLastEntry.flatMap(_.duplicateOf(batch)).foreach { duplicate => + return (updatedProducers, completedTxns.toList, Some(duplicate)) + } val maybeCompletedTxn = updateProducers(batch, updatedProducers, loadingFromLog = false) maybeCompletedTxn.foreach(completedTxns += _) diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 4f53b41df4fc0..61dd0fc47c874 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -450,7 +450,7 @@ private[log] class Cleaner(val id: Int, info("Cleaning segment %s in log %s (largest timestamp %s) into %s, %s deletes." .format(startOffset, log.name, new Date(oldSegmentOpt.largestTimestamp), cleaned.baseOffset, if(retainDeletes) "retaining" else "discarding")) cleanInto(log.topicPartition, oldSegmentOpt, cleaned, map, retainDeletes, log.config.maxMessageSize, transactionMetadata, - log.activeProducers, stats) + log.activeProducersWithLastSequence, stats) currentSegmentOpt = nextSegmentOpt } @@ -503,7 +503,7 @@ private[log] class Cleaner(val id: Int, retainDeletes: Boolean, maxLogMessageSize: Int, transactionMetadata: CleanedTransactionMetadata, - activeProducers: Map[Long, ProducerIdEntry], + activeProducers: Map[Long, Int], stats: CleanerStats) { val logCleanerFilter = new RecordFilter { var discardBatchRecords: Boolean = _ @@ -515,7 +515,7 @@ private[log] class Cleaner(val id: Int, // check if the batch contains the last sequence number for the producer. if so, we cannot // remove the batch just yet or the producer may see an out of sequence error. - if (batch.hasProducerId && activeProducers.get(batch.producerId).exists(_.lastSeq == batch.lastSequence)) + if (batch.hasProducerId && activeProducers.get(batch.producerId).contains(batch.lastSequence)) BatchRetention.RETAIN_EMPTY else if (discardBatchRecords) BatchRetention.DELETE diff --git a/core/src/main/scala/kafka/log/ProducerStateManager.scala b/core/src/main/scala/kafka/log/ProducerStateManager.scala index fc2e34024c32c..4c3d1a18d712d 100644 --- a/core/src/main/scala/kafka/log/ProducerStateManager.scala +++ b/core/src/main/scala/kafka/log/ProducerStateManager.scala @@ -17,7 +17,7 @@ package kafka.log import java.io._ -import java.nio.ByteBuffer +import java.nio.{BufferUnderflowException, ByteBuffer} import java.nio.file.Files import kafka.common.KafkaException @@ -48,33 +48,82 @@ private[log] case class TxnMetadata(producerId: Long, var firstOffset: LogOffset } private[log] object ProducerIdEntry { - val Empty = ProducerIdEntry(RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, - -1, 0, RecordBatch.NO_TIMESTAMP, -1, None) + private[log] val NumBatchesToRetain = 5 + def empty(producerId: Long) = new ProducerIdEntry(producerId, mutable.Queue[BatchMetadata](), RecordBatch.NO_PRODUCER_EPOCH, -1, None) } -private[log] case class ProducerIdEntry(producerId: Long, producerEpoch: Short, lastSeq: Int, lastOffset: Long, - offsetDelta: Int, timestamp: Long, coordinatorEpoch: Int, - currentTxnFirstOffset: Option[Long]) { - def firstSeq: Int = lastSeq - offsetDelta - def firstOffset: Long = lastOffset - offsetDelta +private[log] case class BatchMetadata(lastSeq: Int, lastOffset: Long, offsetDelta: Int, timestamp: Long) { + def firstSeq = lastSeq - offsetDelta + def firstOffset = lastOffset - offsetDelta - def isDuplicate(batch: RecordBatch): Boolean = { - batch.producerEpoch == producerEpoch && - batch.baseSequence == firstSeq && - batch.lastSequence == lastSeq + override def toString: String = { + "BatchMetadata(" + + s"firstSeq=$firstSeq, " + + s"lastSeq=$lastSeq, " + + s"firstOffset=$firstOffset, " + + s"lastOffset=$lastOffset, " + + s"timestamp=$timestamp)" + } +} + +// the batchMetadata is ordered such that the batch with the lowest sequence is at the head of the queue while the +// batch with the highest sequence is at the tail of the queue. We will retain at most ProducerIdEntry.NumBatchesToRetain +// elements in the queue. When the queue is at capacity, we remove the first element to make space for the incoming batch. +private[log] class ProducerIdEntry(val producerId: Long, val batchMetadata: mutable.Queue[BatchMetadata], + var producerEpoch: Short, var coordinatorEpoch: Int, + var currentTxnFirstOffset: Option[Long]) { + + def firstSeq: Int = if (batchMetadata.isEmpty) RecordBatch.NO_SEQUENCE else batchMetadata.front.firstSeq + def firstOffset: Long = if (batchMetadata.isEmpty) -1L else batchMetadata.front.firstOffset + + def lastSeq: Int = if (batchMetadata.isEmpty) RecordBatch.NO_SEQUENCE else batchMetadata.last.lastSeq + def lastDataOffset: Long = if (batchMetadata.isEmpty) -1L else batchMetadata.last.lastOffset + def lastTimestamp = if (batchMetadata.isEmpty) RecordBatch.NO_TIMESTAMP else batchMetadata.last.timestamp + def lastOffsetDelta : Int = if (batchMetadata.isEmpty) 0 else batchMetadata.last.offsetDelta + + def addBatchMetadata(producerEpoch: Short, lastSeq: Int, lastOffset: Long, offsetDelta: Int, timestamp: Long) = { + maybeUpdateEpoch(producerEpoch) + + if (batchMetadata.size == ProducerIdEntry.NumBatchesToRetain) + batchMetadata.dequeue() + + batchMetadata.enqueue(BatchMetadata(lastSeq, lastOffset, offsetDelta, timestamp)) + } + + def maybeUpdateEpoch(producerEpoch: Short): Boolean = { + if (this.producerEpoch != producerEpoch) { + batchMetadata.clear() + this.producerEpoch = producerEpoch + true + } else { + false + } + } + + def removeBatchesOlderThan(offset: Long) = batchMetadata.dropWhile(_.lastOffset < offset) + + def duplicateOf(batch: RecordBatch): Option[BatchMetadata] = { + if (batch.producerEpoch() != producerEpoch) + None + else + batchWithSequenceRange(batch.baseSequence(), batch.lastSequence()) + } + + // Return the batch metadata of the cached batch having the exact sequence range, if any. + def batchWithSequenceRange(firstSeq: Int, lastSeq: Int): Option[BatchMetadata] = { + val duplicate = batchMetadata.filter { case(metadata) => + firstSeq == metadata.firstSeq && lastSeq == metadata.lastSeq + } + duplicate.headOption } override def toString: String = { "ProducerIdEntry(" + s"producerId=$producerId, " + s"producerEpoch=$producerEpoch, " + - s"firstSequence=$firstSeq, " + - s"lastSequence=$lastSeq, " + - s"firstOffset=$firstOffset, " + - s"lastOffset=$lastOffset, " + - s"timestamp=$timestamp, " + s"currentTxnFirstOffset=$currentTxnFirstOffset, " + - s"coordinatorEpoch=$coordinatorEpoch)" + s"coordinatorEpoch=$coordinatorEpoch, " + + s"batchMetadata=$batchMetadata" } } @@ -85,8 +134,10 @@ private[log] case class ProducerIdEntry(producerId: Long, producerEpoch: Short, * as the incoming records are validated. * * @param producerId The id of the producer appending to the log - * @param initialEntry The last entry associated with the producer id. Validation of the first append will be - * based off of this entry initially + * @param currentEntry The current entry associated with the producer id which contains metadata for a fixed number of + * the most recent appends made by the producer. Validation of the first incoming append will + * be made against the lastest append in the current entry. New appends will replace older appends + * in the current entry so that the space overhead is constant. * @param validateSequenceNumbers Whether or not sequence numbers should be validated. The only current use * of this is the consumer offsets topic which uses producer ids from incoming * TxnOffsetCommit, but has no sequence number to validate and does not depend @@ -98,48 +149,46 @@ private[log] case class ProducerIdEntry(producerId: Long, producerEpoch: Short, * retention enforcement. */ private[log] class ProducerAppendInfo(val producerId: Long, - initialEntry: ProducerIdEntry, + currentEntry: ProducerIdEntry, validateSequenceNumbers: Boolean, loadingFromLog: Boolean) { - private var producerEpoch = initialEntry.producerEpoch - private var firstSeq = initialEntry.firstSeq - private var lastSeq = initialEntry.lastSeq - private var lastOffset = initialEntry.lastOffset - private var maxTimestamp = initialEntry.timestamp - private var currentTxnFirstOffset = initialEntry.currentTxnFirstOffset - private var coordinatorEpoch = initialEntry.coordinatorEpoch + private val transactions = ListBuffer.empty[TxnMetadata] private def validateAppend(producerEpoch: Short, firstSeq: Int, lastSeq: Int) = { if (isFenced(producerEpoch)) { throw new ProducerFencedException(s"Producer's epoch is no longer valid. There is probably another producer " + - s"with a newer epoch. $producerEpoch (request epoch), ${this.producerEpoch} (server epoch)") + s"with a newer epoch. $producerEpoch (request epoch), ${currentEntry.producerEpoch} (server epoch)") } else if (validateSequenceNumbers) { - if (producerEpoch != this.producerEpoch) { + if (producerEpoch != currentEntry.producerEpoch) { if (firstSeq != 0) throw new OutOfOrderSequenceException(s"Invalid sequence number for new epoch: $producerEpoch " + s"(request epoch), $firstSeq (seq. number)") - } else if (this.firstSeq == RecordBatch.NO_SEQUENCE && firstSeq != 0) { + } else if (currentEntry.lastSeq == RecordBatch.NO_SEQUENCE && firstSeq != 0) { // the epoch was bumped by a control record, so we expect the sequence number to be reset throw new OutOfOrderSequenceException(s"Out of order sequence number for producerId $producerId: found $firstSeq " + s"(incoming seq. number), but expected 0") - } else if (firstSeq == this.firstSeq && lastSeq == this.lastSeq) { - throw new DuplicateSequenceNumberException(s"Duplicate sequence number for producerId $producerId: (incomingBatch.firstSeq, " + - s"incomingBatch.lastSeq): ($firstSeq, $lastSeq), (lastEntry.firstSeq, lastEntry.lastSeq): " + - s"(${this.firstSeq}, ${this.lastSeq}).") + } else if (isDuplicate(firstSeq, lastSeq)) { + throw new DuplicateSequenceException(s"Duplicate sequence number for producerId $producerId: (incomingBatch.firstSeq, " + + s"incomingBatch.lastSeq): ($firstSeq, $lastSeq).") } else if (!inSequence(firstSeq, lastSeq)) { throw new OutOfOrderSequenceException(s"Out of order sequence number for producerId $producerId: $firstSeq " + - s"(incoming seq. number), ${this.lastSeq} (current end sequence number)") + s"(incoming seq. number), ${currentEntry.lastSeq} (current end sequence number)") } } } + private def isDuplicate(firstSeq: Int, lastSeq: Int): Boolean = { + ((lastSeq != 0 && currentEntry.firstSeq != Int.MaxValue && lastSeq < currentEntry.firstSeq) + || currentEntry.batchWithSequenceRange(firstSeq, lastSeq).isDefined) + } + private def inSequence(firstSeq: Int, lastSeq: Int): Boolean = { - firstSeq == this.lastSeq + 1L || (firstSeq == 0 && this.lastSeq == Int.MaxValue) + firstSeq == currentEntry.lastSeq + 1L || (firstSeq == 0 && currentEntry.lastSeq == Int.MaxValue) } private def isFenced(producerEpoch: Short): Boolean = { - producerEpoch < this.producerEpoch + producerEpoch < currentEntry.producerEpoch } def append(batch: RecordBatch): Option[CompletedTxn] = { @@ -166,18 +215,14 @@ private[log] class ProducerAppendInfo(val producerId: Long, // will generally have removed the beginning entries from each producer id validateAppend(epoch, firstSeq, lastSeq) - this.producerEpoch = epoch - this.firstSeq = firstSeq - this.lastSeq = lastSeq - this.maxTimestamp = lastTimestamp - this.lastOffset = lastOffset + currentEntry.addBatchMetadata(epoch, lastSeq, lastOffset, lastSeq - firstSeq, lastTimestamp) - if (currentTxnFirstOffset.isDefined && !isTransactional) + if (currentEntry.currentTxnFirstOffset.isDefined && !isTransactional) throw new InvalidTxnStateException(s"Expected transactional write from producer $producerId") - if (isTransactional && currentTxnFirstOffset.isEmpty) { + if (isTransactional && currentEntry.currentTxnFirstOffset.isEmpty) { val firstOffset = lastOffset - (lastSeq - firstSeq) - currentTxnFirstOffset = Some(firstOffset) + currentEntry.currentTxnFirstOffset = Some(firstOffset) transactions += new TxnMetadata(producerId, firstOffset) } } @@ -187,44 +232,27 @@ private[log] class ProducerAppendInfo(val producerId: Long, offset: Long, timestamp: Long): CompletedTxn = { if (isFenced(producerEpoch)) - throw new ProducerFencedException(s"Invalid producer epoch: $producerEpoch (zombie): ${this.producerEpoch} (current)") + throw new ProducerFencedException(s"Invalid producer epoch: $producerEpoch (zombie): ${currentEntry.producerEpoch} (current)") - if (this.coordinatorEpoch > endTxnMarker.coordinatorEpoch) + if (currentEntry.coordinatorEpoch > endTxnMarker.coordinatorEpoch) throw new TransactionCoordinatorFencedException(s"Invalid coordinator epoch: ${endTxnMarker.coordinatorEpoch} " + - s"(zombie), $coordinatorEpoch (current)") + s"(zombie), ${currentEntry.coordinatorEpoch} (current)") - if (producerEpoch != this.producerEpoch) { - // it is possible that this control record is the first record seen from a new epoch (the producer - // may fail before sending to the partition or the request itself could fail for some reason). In this - // case, we bump the epoch and reset the sequence numbers - this.producerEpoch = producerEpoch - this.firstSeq = RecordBatch.NO_SEQUENCE - this.lastSeq = RecordBatch.NO_SEQUENCE - } else { - // the control record is the last append to the log, so the last offset will be updated to point to it. - // However, the sequence numbers still point to the previous batch, so the duplicate check would no longer - // be correct: it would return the wrong offset. To fix this, we treat the control record as a batch - // of size 1 which uses the last appended sequence number. - this.firstSeq = this.lastSeq - } + currentEntry.maybeUpdateEpoch(producerEpoch) - val firstOffset = currentTxnFirstOffset match { + val firstOffset = currentEntry.currentTxnFirstOffset match { case Some(txnFirstOffset) => txnFirstOffset case None => transactions += new TxnMetadata(producerId, offset) offset } - this.lastOffset = offset - this.currentTxnFirstOffset = None - this.maxTimestamp = timestamp - this.coordinatorEpoch = endTxnMarker.coordinatorEpoch + currentEntry.currentTxnFirstOffset = None + currentEntry.coordinatorEpoch = endTxnMarker.coordinatorEpoch CompletedTxn(producerId, firstOffset, offset, endTxnMarker.controlType == ControlRecordType.ABORT) } - def lastEntry: ProducerIdEntry = - ProducerIdEntry(producerId, producerEpoch, lastSeq, lastOffset, lastSeq - firstSeq, maxTimestamp, - coordinatorEpoch, currentTxnFirstOffset) + def latestEntry: ProducerIdEntry = currentEntry def startedTransactions: List[TxnMetadata] = transactions.toList @@ -243,11 +271,11 @@ private[log] class ProducerAppendInfo(val producerId: Long, override def toString: String = { "ProducerAppendInfo(" + s"producerId=$producerId, " + - s"producerEpoch=$producerEpoch, " + - s"firstSequence=$firstSeq, " + - s"lastSequence=$lastSeq, " + - s"currentTxnFirstOffset=$currentTxnFirstOffset, " + - s"coordinatorEpoch=$coordinatorEpoch, " + + s"producerEpoch=${currentEntry.producerEpoch}, " + + s"firstSequence=${currentEntry.firstSeq}, " + + s"lastSequence=${currentEntry.lastSeq}, " + + s"currentTxnFirstOffset=${currentEntry.currentTxnFirstOffset}, " + + s"coordinatorEpoch=${currentEntry.coordinatorEpoch}, " + s"startedTransactions=$transactions)" } } @@ -309,7 +337,7 @@ object ProducerStateManager { val offsetDelta = producerEntryStruct.getInt(OffsetDeltaField) val coordinatorEpoch = producerEntryStruct.getInt(CoordinatorEpochField) val currentTxnFirstOffset = producerEntryStruct.getLong(CurrentTxnFirstOffsetField) - val newEntry = ProducerIdEntry(producerId, producerEpoch, seq, offset, offsetDelta, timestamp, + val newEntry = new ProducerIdEntry(producerId, mutable.Queue[BatchMetadata](BatchMetadata(seq, offset, offsetDelta, timestamp)), producerEpoch, coordinatorEpoch, if (currentTxnFirstOffset >= 0) Some(currentTxnFirstOffset) else None) newEntry } @@ -329,9 +357,9 @@ object ProducerStateManager { producerEntryStruct.set(ProducerIdField, producerId) .set(ProducerEpochField, entry.producerEpoch) .set(LastSequenceField, entry.lastSeq) - .set(LastOffsetField, entry.lastOffset) - .set(OffsetDeltaField, entry.offsetDelta) - .set(TimestampField, entry.timestamp) + .set(LastOffsetField, entry.lastDataOffset) + .set(OffsetDeltaField, entry.lastOffsetDelta) + .set(TimestampField, entry.lastTimestamp) .set(CoordinatorEpochField, entry.coordinatorEpoch) .set(CurrentTxnFirstOffsetField, entry.currentTxnFirstOffset.getOrElse(-1L)) producerEntryStruct @@ -472,7 +500,7 @@ class ProducerStateManager(val topicPartition: TopicPartition, } private def isProducerExpired(currentTimeMs: Long, producerIdEntry: ProducerIdEntry): Boolean = - producerIdEntry.currentTxnFirstOffset.isEmpty && currentTimeMs - producerIdEntry.timestamp >= maxProducerIdExpirationMs + producerIdEntry.currentTxnFirstOffset.isEmpty && currentTimeMs - producerIdEntry.lastTimestamp >= maxProducerIdExpirationMs /** * Expire any producer ids which have been idle longer than the configured maximum expiration timeout. @@ -508,7 +536,7 @@ class ProducerStateManager(val topicPartition: TopicPartition, } def prepareUpdate(producerId: Long, loadingFromLog: Boolean): ProducerAppendInfo = - new ProducerAppendInfo(producerId, lastEntry(producerId).getOrElse(ProducerIdEntry.Empty), validateSequenceNumbers, + new ProducerAppendInfo(producerId, lastEntry(producerId).getOrElse(ProducerIdEntry.empty(producerId)), validateSequenceNumbers, loadingFromLog) /** @@ -520,7 +548,7 @@ class ProducerStateManager(val topicPartition: TopicPartition, trace(s"Updated producer ${appendInfo.producerId} state to $appendInfo") - val entry = appendInfo.lastEntry + val entry = appendInfo.latestEntry producers.put(appendInfo.producerId, entry) appendInfo.startedTransactions.foreach { txn => ongoingTxns.put(txn.firstOffset.messageOffset, txn) @@ -562,7 +590,8 @@ class ProducerStateManager(val topicPartition: TopicPartition, def oldestSnapshotOffset: Option[Long] = oldestSnapshotFile.map(file => offsetFromFilename(file.getName)) private def isProducerRetained(producerIdEntry: ProducerIdEntry, logStartOffset: Long): Boolean = { - producerIdEntry.lastOffset >= logStartOffset + producerIdEntry.removeBatchesOlderThan(logStartOffset) + producerIdEntry.lastDataOffset >= logStartOffset } /** diff --git a/core/src/main/scala/kafka/tools/DumpLogSegments.scala b/core/src/main/scala/kafka/tools/DumpLogSegments.scala index 025617fc6b3ea..c4f7ce0218b08 100755 --- a/core/src/main/scala/kafka/tools/DumpLogSegments.scala +++ b/core/src/main/scala/kafka/tools/DumpLogSegments.scala @@ -115,7 +115,7 @@ object DumpLogSegments { case Log.TimeIndexFileSuffix => dumpTimeIndex(file, indexSanityOnly, verifyOnly, timeIndexDumpErrors, maxMessageSize) case Log.PidSnapshotFileSuffix => - dumpPidSnapshot(file) + dumpProducerIdSnapshot(file) case Log.TxnIndexFileSuffix => dumpTxnIndex(file) case _ => @@ -152,12 +152,12 @@ object DumpLogSegments { } } - private def dumpPidSnapshot(file: File): Unit = { + private def dumpProducerIdSnapshot(file: File): Unit = { try { - ProducerStateManager.readSnapshot(file).foreach { entry=> - println(s"producerId: ${entry.producerId} producerEpoch: ${entry.producerEpoch} lastSequence: ${entry.lastSeq} " + - s"lastOffset: ${entry.lastOffset} offsetDelta: ${entry.offsetDelta} lastTimestamp: ${entry.timestamp} " + - s"coordinatorEpoch: ${entry.coordinatorEpoch} currentTxnFirstOffset: ${entry.currentTxnFirstOffset}") + ProducerStateManager.readSnapshot(file).foreach { entry => + println(s"producerId: ${entry.producerId} producerEpoch: ${entry.producerEpoch} " + + s"coordinatorEpoch: ${entry.coordinatorEpoch} currentTxnFirstOffset: ${entry.currentTxnFirstOffset} " + + s"cachedMetadata: ${entry.batchMetadata}") } } catch { case e: CorruptSnapshotException => diff --git a/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala b/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala index a11972eec3dd3..1bde7b1211093 100644 --- a/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala +++ b/core/src/test/scala/integration/kafka/api/ProducerBounceTest.scala @@ -122,7 +122,7 @@ class ProducerBounceTest extends KafkaServerTestHarness { val producerConfig = new Properties() producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") - producerConfig.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1") + producerConfig.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5") val producerConfigWithCompression = new Properties() producerConfigWithCompression ++= producerConfig producerConfigWithCompression.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4") diff --git a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala index 810f48192f8f5..1a85d340c634d 100644 --- a/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala +++ b/core/src/test/scala/integration/kafka/api/TransactionsBounceTest.scala @@ -24,12 +24,15 @@ import kafka.server.KafkaConfig import kafka.utils.{ShutdownableThread, TestUtils} import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback +import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.protocol.SecurityProtocol import org.junit.Test import scala.collection.JavaConverters._ import org.junit.Assert._ +import scala.collection.mutable + class TransactionsBounceTest extends KafkaServerTestHarness { private val producerBufferSize = 65536 @@ -76,12 +79,12 @@ class TransactionsBounceTest extends KafkaServerTestHarness { // basic idea is to seed a topic with 10000 records, and copy it transactionally while bouncing brokers // constantly through the period. val consumerGroup = "myGroup" - val numInputRecords = 5000 + val numInputRecords = 10000 createTopics() TestUtils.seedTopicWithNumberedRecords(inputTopic, numInputRecords, servers) val consumer = createConsumerAndSubscribeToTopics(consumerGroup, List(inputTopic)) - val producer = TestUtils.createTransactionalProducer("test-txn", servers) + val producer = TestUtils.createTransactionalProducer("test-txn", servers, 512) producer.initTransactions() @@ -125,9 +128,20 @@ class TransactionsBounceTest extends KafkaServerTestHarness { scheduler.shutdown() val verifyingConsumer = createConsumerAndSubscribeToTopics("randomGroup", List(outputTopic), readCommitted = true) - val outputRecords = TestUtils.pollUntilAtLeastNumRecords(verifyingConsumer, numInputRecords).map { record => - TestUtils.assertCommittedAndGetValue(record).toInt + val recordsByPartition = new mutable.HashMap[TopicPartition, mutable.ListBuffer[Int]]() + TestUtils.pollUntilAtLeastNumRecords(verifyingConsumer, numInputRecords).foreach { record => + val value = TestUtils.assertCommittedAndGetValue(record).toInt + val topicPartition = new TopicPartition(record.topic(), record.partition()) + recordsByPartition.getOrElseUpdate(topicPartition, new mutable.ListBuffer[Int]) + .append(value) + } + + val outputRecords = new mutable.ListBuffer[Int]() + recordsByPartition.values.foreach { case (partitionValues) => + assertEquals("Out of order messages detected", partitionValues, partitionValues.sorted) + outputRecords.appendAll(partitionValues) } + val recordSet = outputRecords.toSet assertEquals(numInputRecords, recordSet.size) diff --git a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala index 2a07532abaaa3..61a8492db558f 100644 --- a/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala @@ -17,6 +17,7 @@ package kafka.log import java.io.File +import java.util.concurrent.ConcurrentLinkedDeque import kafka.utils.TestUtils import kafka.utils.TestUtils.checkEquals @@ -317,7 +318,7 @@ class LogSegmentTest { // recover again, but this time assuming the transaction from pid2 began on a previous segment stateManager = new ProducerStateManager(topicPartition, logDir) - stateManager.loadProducerEntry(ProducerIdEntry(pid2, producerEpoch, 10, 90L, 5, RecordBatch.NO_TIMESTAMP, 0, Some(75L))) + stateManager.loadProducerEntry(new ProducerIdEntry(pid2, mutable.Queue[BatchMetadata](BatchMetadata(10, 90L, 5, RecordBatch.NO_TIMESTAMP)), producerEpoch, 0, Some(75L))) segment.recover(stateManager) assertEquals(108L, stateManager.mapEndOffset) diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala index ad64e39a0a57c..2ae62c5338051 100755 --- a/core/src/test/scala/unit/kafka/log/LogTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogTest.scala @@ -418,14 +418,11 @@ class LogTest { log.truncateTo(baseOffset + 4) - val activeProducers = log.activeProducers + val activeProducers = log.activeProducersWithLastSequence assertTrue(activeProducers.contains(pid)) - val entry = activeProducers(pid) - assertEquals(0, entry.firstSeq) - assertEquals(baseOffset, entry.firstOffset) - assertEquals(3, entry.lastSeq) - assertEquals(baseOffset + 3, entry.lastOffset) + val lastSeq = activeProducers(pid) + assertEquals(3, lastSeq) } @Test @@ -462,14 +459,11 @@ class LogTest { log.truncateTo(baseOffset + 2) - val activeProducers = log.activeProducers + val activeProducers = log.activeProducersWithLastSequence assertTrue(activeProducers.contains(pid)) - val entry = activeProducers(pid) - assertEquals(0, entry.firstSeq) - assertEquals(baseOffset, entry.firstOffset) - assertEquals(1, entry.lastSeq) - assertEquals(baseOffset + 1, entry.lastOffset) + val lastSeq = activeProducers(pid) + assertEquals(1, lastSeq) } @Test @@ -498,14 +492,11 @@ class LogTest { val filteredRecords = MemoryRecords.readableRecords(filtered) log.appendAsFollower(filteredRecords) - val activeProducers = log.activeProducers + val activeProducers = log.activeProducersWithLastSequence assertTrue(activeProducers.contains(pid)) - val entry = activeProducers(pid) - assertEquals(0, entry.firstSeq) - assertEquals(baseOffset, entry.firstOffset) - assertEquals(3, entry.lastSeq) - assertEquals(baseOffset + 3, entry.lastOffset) + val lastSeq = activeProducers(pid) + assertEquals(3, lastSeq) } @Test @@ -547,13 +538,13 @@ class LogTest { } log.truncateTo(1L) - assertEquals(1, log.activeProducers.size) + assertEquals(1, log.activeProducersWithLastSequence.size) - val pidEntryOpt = log.activeProducers.get(pid) - assertTrue(pidEntryOpt.isDefined) + val lastSeqOpt = log.activeProducersWithLastSequence.get(pid) + assertTrue(lastSeqOpt.isDefined) - val pidEntry = pidEntryOpt.get - assertEquals(0, pidEntry.lastSeq) + val lastSeq = lastSeqOpt.get + assertEquals(0, lastSeq) } @Test @@ -568,21 +559,21 @@ class LogTest { producerEpoch = epoch, sequence = 0), leaderEpoch = 0) log.appendAsLeader(TestUtils.records(List(new SimpleRecord(mockTime.milliseconds(), "b".getBytes)), producerId = pid2, producerEpoch = epoch, sequence = 0), leaderEpoch = 0) - assertEquals(2, log.activeProducers.size) + assertEquals(2, log.activeProducersWithLastSequence.size) log.maybeIncrementLogStartOffset(1L) - assertEquals(1, log.activeProducers.size) - val retainedEntryOpt = log.activeProducers.get(pid2) - assertTrue(retainedEntryOpt.isDefined) - assertEquals(0, retainedEntryOpt.get.lastSeq) + assertEquals(1, log.activeProducersWithLastSequence.size) + val retainedLastSeqOpt = log.activeProducersWithLastSequence.get(pid2) + assertTrue(retainedLastSeqOpt.isDefined) + assertEquals(0, retainedLastSeqOpt.get) log.close() val reloadedLog = createLog(logDir, logConfig, logStartOffset = 1L) - assertEquals(1, reloadedLog.activeProducers.size) - val reloadedEntryOpt = log.activeProducers.get(pid2) - assertEquals(retainedEntryOpt, reloadedEntryOpt) + assertEquals(1, reloadedLog.activeProducersWithLastSequence.size) + val reloadedLastSeqOpt = log.activeProducersWithLastSequence.get(pid2) + assertEquals(retainedLastSeqOpt, reloadedLastSeqOpt) } @Test @@ -600,24 +591,24 @@ class LogTest { producerEpoch = epoch, sequence = 0), leaderEpoch = 0) assertEquals(2, log.logSegments.size) - assertEquals(2, log.activeProducers.size) + assertEquals(2, log.activeProducersWithLastSequence.size) log.maybeIncrementLogStartOffset(1L) log.onHighWatermarkIncremented(log.logEndOffset) log.deleteOldSegments() assertEquals(1, log.logSegments.size) - assertEquals(1, log.activeProducers.size) - val retainedEntryOpt = log.activeProducers.get(pid2) - assertTrue(retainedEntryOpt.isDefined) - assertEquals(0, retainedEntryOpt.get.lastSeq) + assertEquals(1, log.activeProducersWithLastSequence.size) + val retainedLastSeqOpt = log.activeProducersWithLastSequence.get(pid2) + assertTrue(retainedLastSeqOpt.isDefined) + assertEquals(0, retainedLastSeqOpt.get) log.close() val reloadedLog = createLog(logDir, logConfig, logStartOffset = 1L) - assertEquals(1, reloadedLog.activeProducers.size) - val reloadedEntryOpt = log.activeProducers.get(pid2) - assertEquals(retainedEntryOpt, reloadedEntryOpt) + assertEquals(1, reloadedLog.activeProducersWithLastSequence.size) + val reloadedEntryOpt = log.activeProducersWithLastSequence.get(pid2) + assertEquals(retainedLastSeqOpt, reloadedEntryOpt) } @Test @@ -659,13 +650,13 @@ class LogTest { log.takeProducerSnapshot() assertEquals(3, log.logSegments.size) - assertEquals(Set(pid1, pid2), log.activeProducers.keySet) + assertEquals(Set(pid1, pid2), log.activeProducersWithLastSequence.keySet) log.onHighWatermarkIncremented(log.logEndOffset) log.deleteOldSegments() assertEquals(2, log.logSegments.size) - assertEquals(Set(pid2), log.activeProducers.keySet) + assertEquals(Set(pid2), log.activeProducersWithLastSequence.keySet) } @Test @@ -749,13 +740,13 @@ class LogTest { val records = Seq(new SimpleRecord(mockTime.milliseconds(), "foo".getBytes)) log.appendAsLeader(TestUtils.records(records, producerId = pid, producerEpoch = 0, sequence = 0), leaderEpoch = 0) - assertEquals(Set(pid), log.activeProducers.keySet) + assertEquals(Set(pid), log.activeProducersWithLastSequence.keySet) mockTime.sleep(producerIdExpirationCheckIntervalMs) - assertEquals(Set(pid), log.activeProducers.keySet) + assertEquals(Set(pid), log.activeProducersWithLastSequence.keySet) mockTime.sleep(producerIdExpirationCheckIntervalMs) - assertEquals(Set(), log.activeProducers.keySet) + assertEquals(Set(), log.activeProducersWithLastSequence.keySet) } @Test @@ -805,16 +796,22 @@ class LogTest { case _: OutOfOrderSequenceException => // Good! } - // Append a Duplicate of an entry in the middle of the log. This is not allowed. + // Append a duplicate of the batch which is 4th from the tail. This should succeed without error since we + // retain the batch metadata of the last 5 batches. + val duplicateOfFourth = TestUtils.records(List(new SimpleRecord(mockTime.milliseconds, "key".getBytes, "value".getBytes)), + producerId = pid, producerEpoch = epoch, sequence = 2) + log.appendAsLeader(duplicateOfFourth, leaderEpoch = 0) + + // Append a Duplicate of an entry older than the last 5 appended batches. This should result in a DuplicateSequenceNumberException. try { val records = TestUtils.records( List(new SimpleRecord(mockTime.milliseconds, s"key-1".getBytes, s"value-1".getBytes)), producerId = pid, producerEpoch = epoch, sequence = 1) log.appendAsLeader(records, leaderEpoch = 0) - fail ("Should have received an OutOfOrderSequenceException since we attempted to append a duplicate of a records " + - "in the middle of the log.") + fail ("Should have received an DuplicateSequenceNumberException since we attempted to append a duplicate of a batch" + + "which is older than the last 5 appended batches.") } catch { - case _: OutOfOrderSequenceException => // Good! + case _: DuplicateSequenceException => // Good! } // Append a duplicate entry with a single records at the tail of the log. This should return the appendInfo of the original entry. @@ -872,7 +869,7 @@ class LogTest { } } - @Test(expected = classOf[DuplicateSequenceNumberException]) + @Test(expected = classOf[DuplicateSequenceException]) def testDuplicateAppendToFollower() : Unit = { val logConfig = createLogConfig(segmentBytes = 1024 * 1024 * 5) val log = createLog(logDir, logConfig) @@ -888,7 +885,7 @@ class LogTest { partitionLeaderEpoch, new SimpleRecord("a".getBytes), new SimpleRecord("b".getBytes))) } - @Test(expected = classOf[DuplicateSequenceNumberException]) + @Test(expected = classOf[DuplicateSequenceException]) def testMultipleProducersWithDuplicatesInSingleAppend() : Unit = { val logConfig = createLogConfig(segmentBytes = 1024 * 1024 * 5) val log = createLog(logDir, logConfig) diff --git a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala index bb41380160287..976bbd7e92148 100644 --- a/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala +++ b/core/src/test/scala/unit/kafka/log/ProducerStateManagerTest.scala @@ -63,7 +63,7 @@ class ProducerStateManagerTest extends JUnitSuite { append(stateManager, producerId, epoch, 1, 0L, 1L) // Duplicate sequence number (matches previous sequence number) - assertThrows[DuplicateSequenceNumberException] { + assertThrows[DuplicateSequenceException] { append(stateManager, producerId, epoch, 1, 0L, 1L) } @@ -95,7 +95,8 @@ class ProducerStateManagerTest extends JUnitSuite { val lastEntry = maybeLastEntry.get assertEquals(epoch, lastEntry.producerEpoch) - assertEquals(0, lastEntry.firstSeq) + + assertEquals(Int.MaxValue, lastEntry.firstSeq) assertEquals(0, lastEntry.lastSeq) } @@ -122,7 +123,7 @@ class ProducerStateManagerTest extends JUnitSuite { assertEquals(epoch, lastEntry.producerEpoch) assertEquals(sequence, lastEntry.firstSeq) assertEquals(sequence, lastEntry.lastSeq) - assertEquals(offset, lastEntry.lastOffset) + assertEquals(offset, lastEntry.lastDataOffset) assertEquals(offset, lastEntry.firstOffset) } @@ -158,7 +159,7 @@ class ProducerStateManagerTest extends JUnitSuite { val producerEpoch = 0.toShort val offset = 992342L val seq = 0 - val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerIdEntry.Empty, validateSequenceNumbers = true, + val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerIdEntry.empty(producerId), validateSequenceNumbers = true, loadingFromLog = false) producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, isTransactional = true) @@ -175,7 +176,7 @@ class ProducerStateManagerTest extends JUnitSuite { val producerEpoch = 0.toShort val offset = 992342L val seq = 0 - val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerIdEntry.Empty, validateSequenceNumbers = true, + val producerAppendInfo = new ProducerAppendInfo(producerId, ProducerIdEntry.empty(producerId), validateSequenceNumbers = true, loadingFromLog = false) producerAppendInfo.append(producerEpoch, seq, seq, time.milliseconds(), offset, isTransactional = true) @@ -198,22 +199,22 @@ class ProducerStateManagerTest extends JUnitSuite { val appendInfo = stateManager.prepareUpdate(producerId, loadingFromLog = false) appendInfo.append(producerEpoch, 1, 5, time.milliseconds(), 20L, isTransactional = true) - var lastEntry = appendInfo.lastEntry + var lastEntry = appendInfo.latestEntry assertEquals(producerEpoch, lastEntry.producerEpoch) - assertEquals(1, lastEntry.firstSeq) + assertEquals(0, lastEntry.firstSeq) assertEquals(5, lastEntry.lastSeq) - assertEquals(16L, lastEntry.firstOffset) - assertEquals(20L, lastEntry.lastOffset) + assertEquals(9L, lastEntry.firstOffset) + assertEquals(20L, lastEntry.lastDataOffset) assertEquals(Some(16L), lastEntry.currentTxnFirstOffset) assertEquals(List(new TxnMetadata(producerId, 16L)), appendInfo.startedTransactions) appendInfo.append(producerEpoch, 6, 10, time.milliseconds(), 30L, isTransactional = true) - lastEntry = appendInfo.lastEntry + lastEntry = appendInfo.latestEntry assertEquals(producerEpoch, lastEntry.producerEpoch) - assertEquals(6, lastEntry.firstSeq) + assertEquals(0, lastEntry.firstSeq) assertEquals(10, lastEntry.lastSeq) - assertEquals(26L, lastEntry.firstOffset) - assertEquals(30L, lastEntry.lastOffset) + assertEquals(9L, lastEntry.firstOffset) + assertEquals(30L, lastEntry.lastDataOffset) assertEquals(Some(16L), lastEntry.currentTxnFirstOffset) assertEquals(List(new TxnMetadata(producerId, 16L)), appendInfo.startedTransactions) @@ -224,12 +225,13 @@ class ProducerStateManagerTest extends JUnitSuite { assertEquals(40L, completedTxn.lastOffset) assertFalse(completedTxn.isAborted) - lastEntry = appendInfo.lastEntry + lastEntry = appendInfo.latestEntry assertEquals(producerEpoch, lastEntry.producerEpoch) - assertEquals(10, lastEntry.firstSeq) + // verify that appending the transaction marker doesn't affect the metadata of the cached record batches. + assertEquals(0, lastEntry.firstSeq) assertEquals(10, lastEntry.lastSeq) - assertEquals(40L, lastEntry.firstOffset) - assertEquals(40L, lastEntry.lastOffset) + assertEquals(9L, lastEntry.firstOffset) + assertEquals(30L, lastEntry.lastDataOffset) assertEquals(coordinatorEpoch, lastEntry.coordinatorEpoch) assertEquals(None, lastEntry.currentTxnFirstOffset) assertEquals(List(new TxnMetadata(producerId, 16L)), appendInfo.startedTransactions) @@ -421,7 +423,7 @@ class ProducerStateManagerTest extends JUnitSuite { val maybeEntry = stateManager.lastEntry(anotherPid) assertTrue(maybeEntry.isDefined) - assertEquals(3L, maybeEntry.get.lastOffset) + assertEquals(3L, maybeEntry.get.lastDataOffset) stateManager.truncateHead(3) assertEquals(Set(anotherPid), stateManager.activeProducers.keySet) @@ -452,7 +454,7 @@ class ProducerStateManagerTest extends JUnitSuite { val entry = stateManager.lastEntry(pid2) assertTrue(entry.isDefined) assertEquals(0, entry.get.lastSeq) - assertEquals(1L, entry.get.lastOffset) + assertEquals(1L, entry.get.lastDataOffset) } @Test @@ -663,7 +665,7 @@ class ProducerStateManagerTest extends JUnitSuite { assertFalse(snapshotToTruncate.exists()) val loadedProducerState = reloadedStateManager.activeProducers(producerId) - assertEquals(0L, loadedProducerState.lastOffset) + assertEquals(0L, loadedProducerState.lastDataOffset) } private def appendEndTxnMarker(mapping: ProducerStateManager, diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index a52c83c11fed8..902d1c3baccdd 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -1379,11 +1379,12 @@ object TestUtils extends Logging { records } - def createTransactionalProducer(transactionalId: String, servers: Seq[KafkaServer]) = { + def createTransactionalProducer(transactionalId: String, servers: Seq[KafkaServer], batchSize: Int = 16384) = { val props = new Properties() props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId) - props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1") + props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5") props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") + props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize.toString) TestUtils.createNewProducer(TestUtils.getBrokerListStrFromServers(servers), retries = Integer.MAX_VALUE, acks = -1, props = Some(props)) } diff --git a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java index 3903a3a8fae00..0d74645379ebe 100644 --- a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java +++ b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java @@ -149,6 +149,11 @@ private static KafkaProducer createProducer(Namespace parsedArgs "org.apache.kafka.common.serialization.StringSerializer"); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"); + // We set a small batch size to ensure that we have multiple inflight requests per transaction. + // If it is left at the default, each transaction will have only one batch per partition, hence not testing + // the case with multiple inflights. + props.put(ProducerConfig.BATCH_SIZE_CONFIG, "512"); + props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); return new KafkaProducer<>(props); } @@ -252,7 +257,6 @@ public static void main(String[] args) throws IOException { maxMessages = Math.min(messagesRemaining(consumer, inputPartition), maxMessages); final boolean enableRandomAborts = parsedArgs.getBoolean("enableRandomAborts"); - producer.initTransactions(); final AtomicBoolean isShuttingDown = new AtomicBoolean(false);