From 588e26b16b9a0eb4ac2b625b41a56a10c72bc5c1 Mon Sep 17 00:00:00 2001 From: Sumant Tambe Date: Wed, 20 Dec 2017 11:22:52 -0800 Subject: [PATCH] Rebasing KIP-91 delivery.timeout.ms for kafka 1.1.0 Avoiding overflow when deliveryTimeoutMs is MAX_VALUE per-partition map for tracking soon to expire batches Updated tests --- .../kafka/clients/producer/KafkaProducer.java | 52 ++-- .../clients/producer/ProducerConfig.java | 9 +- .../producer/internals/ProducerBatch.java | 95 +++++--- .../producer/internals/RecordAccumulator.java | 181 ++++++++++---- .../clients/producer/internals/Sender.java | 4 +- .../apache/kafka/common/config/ConfigDef.java | 12 +- .../producer/internals/ProducerBatchTest.java | 32 +-- .../internals/RecordAccumulatorTest.java | 228 ++++++++++++------ .../producer/internals/SenderTest.java | 222 ++++++++++++++--- .../internals/TransactionManagerTest.java | 6 +- .../apache/kafka/connect/runtime/Worker.java | 1 + .../scala/unit/kafka/utils/TestUtils.scala | 4 + 12 files changed, 617 insertions(+), 229 deletions(-) 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 a5af5b60093d9..4467271949d56 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 @@ -16,6 +16,17 @@ */ package org.apache.kafka.clients.producer; +import java.net.InetSocketAddress; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Properties; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.ClientUtils; import org.apache.kafka.clients.KafkaClient; @@ -24,6 +35,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.consumer.OffsetCommitCallback; +import org.apache.kafka.clients.producer.internals.BufferPool; import org.apache.kafka.clients.producer.internals.ProducerInterceptors; import org.apache.kafka.clients.producer.internals.ProducerMetrics; import org.apache.kafka.clients.producer.internals.RecordAccumulator; @@ -68,18 +80,6 @@ import org.apache.kafka.common.utils.Time; import org.slf4j.Logger; -import java.net.InetSocketAddress; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Objects; -import java.util.Properties; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; - import static org.apache.kafka.common.serialization.ExtendedSerializer.Wrapper.ensureExtended; /** @@ -234,6 +234,7 @@ public class KafkaProducer implements Producer { private static final AtomicInteger PRODUCER_CLIENT_ID_SEQUENCE = new AtomicInteger(1); private static final String JMX_PREFIX = "kafka.producer"; public static final String NETWORK_THREAD_PREFIX = "kafka-producer-network-thread"; + public static final String PRODUCER_METRIC_GROUP_NAME = "producer-metrics"; private final String clientId; // Visible for testing @@ -391,18 +392,22 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali int retries = configureRetries(config, transactionManager != null, log); int maxInflightRequests = configureInflightRequests(config, transactionManager != null); short acks = configureAcks(config, transactionManager != null, log); + long deliveryTimeoutMs = configureDeliveryTimeout(config); this.apiVersions = new ApiVersions(); this.accumulator = new RecordAccumulator(logContext, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), - this.totalMemorySize, this.compressionType, config.getLong(ProducerConfig.LINGER_MS_CONFIG), retryBackoffMs, + this.requestTimeoutMs, + deliveryTimeoutMs, metrics, + PRODUCER_METRIC_GROUP_NAME, time, apiVersions, - transactionManager); + transactionManager, + new BufferPool(this.totalMemorySize, config.getInt(ProducerConfig.BATCH_SIZE_CONFIG), metrics, time, PRODUCER_METRIC_GROUP_NAME)); List addresses = ClientUtils.parseAndValidateAddresses(config.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG)); if (metadata != null) { this.metadata = metadata; @@ -458,6 +463,25 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali } } + private static long configureDeliveryTimeout(ProducerConfig config) { + long deliveryTimeoutMs = config.getLong(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG); + long lingerMs = config.getLong(ProducerConfig.LINGER_MS_CONFIG); + int requestTimeoutMs = config.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG); + + // when the sum of lingerMs and requestTimeoutMs overflows a long we make sure deliveryTimeoutMs is at least + // equal to lingerMs. + boolean overflow = lingerMs + requestTimeoutMs < 0L; + boolean invalid = overflow ? deliveryTimeoutMs < lingerMs : deliveryTimeoutMs < lingerMs + requestTimeoutMs; + + if (invalid) { + throw new ConfigException("Must set " + ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG + " higher than " + + ProducerConfig.LINGER_MS_CONFIG + " + " + + ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG); + } + + return deliveryTimeoutMs; + } + private static TransactionManager configureTransactionState(ProducerConfig config, LogContext logContext, Logger log) { TransactionManager transactionManager = null; diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java index 0c484dfae05a1..a8f1aa173ce50 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java @@ -99,6 +99,12 @@ public class ProducerConfig extends AbstractConfig { + "specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting " + LINGER_MS_CONFIG + "=5, " + "for example, would have the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absence of load."; + /** delivery.timeout.ms */ + public static final String DELIVERY_TIMEOUT_MS_CONFIG = "delivery.timeout.ms"; + private static final String DELIVERY_TIMEOUT_MS_DOC = "An upper bound on the time to report success or failure after Producer.send() returns. " + + "Producer may report failure to send a message earlier than this config if all the retries are exhausted or " + + "a record is added to a batch nearing expiration."; + /** client.id */ public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG; @@ -224,7 +230,7 @@ public class ProducerConfig extends AbstractConfig { static { CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, Collections.emptyList(), new ConfigDef.NonNullValidator(), Importance.HIGH, CommonClientConfigs.BOOTSTRAP_SERVERS_DOC) .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC) - .define(RETRIES_CONFIG, Type.INT, 0, between(0, Integer.MAX_VALUE), Importance.HIGH, RETRIES_DOC) + .define(RETRIES_CONFIG, Type.INT, Integer.MAX_VALUE, between(0, Integer.MAX_VALUE), Importance.HIGH, RETRIES_DOC) .define(ACKS_CONFIG, Type.STRING, "1", @@ -234,6 +240,7 @@ public class ProducerConfig extends AbstractConfig { .define(COMPRESSION_TYPE_CONFIG, Type.STRING, "none", Importance.HIGH, COMPRESSION_TYPE_DOC) .define(BATCH_SIZE_CONFIG, Type.INT, 16384, atLeast(0), Importance.MEDIUM, BATCH_SIZE_DOC) .define(LINGER_MS_CONFIG, Type.LONG, 0, atLeast(0L), Importance.MEDIUM, LINGER_MS_DOC) + .define(DELIVERY_TIMEOUT_MS_CONFIG, Type.LONG, 120 * 1000, atLeast(0L), Importance.MEDIUM, DELIVERY_TIMEOUT_MS_DOC) .define(CLIENT_ID_CONFIG, Type.STRING, "", Importance.MEDIUM, CommonClientConfigs.CLIENT_ID_DOC) .define(SEND_BUFFER_CONFIG, Type.INT, 128 * 1024, atLeast(-1), Importance.MEDIUM, CommonClientConfigs.SEND_BUFFER_DOC) .define(RECEIVE_BUFFER_CONFIG, Type.INT, 32 * 1024, atLeast(-1), Importance.MEDIUM, CommonClientConfigs.RECEIVE_BUFFER_DOC) 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 ea0f0f7dff953..fc3b99d060142 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 @@ -77,13 +77,13 @@ private enum FinalState { ABORTED, FAILED, SUCCEEDED } private boolean retry; private boolean reopened = false; - public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long now) { - this(tp, recordsBuilder, now, false); + public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long creationTime) { + this(tp, recordsBuilder, creationTime, false); } - public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long now, boolean isSplitBatch) { - this.createdMs = now; - this.lastAttemptMs = now; + public ProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long creationTime, boolean isSplitBatch) { + this.createdMs = creationTime; + this.lastAttemptMs = creationTime; this.recordsBuilder = recordsBuilder; this.topicPartition = tp; this.lastAppendTime = createdMs; @@ -158,7 +158,17 @@ public void abort(RuntimeException exception) { } /** - * Complete the request. If the batch was previously aborted, this is a no-op. + * Finalize the state of a batch. Final state, once set, is immutable. This function may be called + * once or twice on a batch. It may be called twice if + * 1. An inflight batch expires before a response from the broker is received. The batch's final + * state is set to FAILED. But it could succeed on the broker and second time around batch.done() may + * try to set SUCCEEDED final state. + * 2. If a transaction abortion happens or if the producer is closed forcefully, the final state is + * ABORTED but again it could succeed if broker responds with a success. + * + * Attempted transitions from [FAILED | ABORTED] --> SUCCEEDED are logged. + * Attempted transitions from one failure state to the same or a different failed state are ignored. + * Attempted transitions from SUCCEEDED to the same or a failed state throw an exception. * * @param baseOffset The base offset of the messages assigned by the server * @param logAppendTime The log append time or -1 if CreateTime is being used @@ -166,26 +176,32 @@ public void abort(RuntimeException exception) { * @return true if the batch was completed successfully and false if the batch was previously aborted */ public boolean done(long baseOffset, long logAppendTime, RuntimeException exception) { - final FinalState finalState; - if (exception == null) { + final FinalState tryFinalState = (exception == null) ? FinalState.SUCCEEDED : FinalState.FAILED; + + if (tryFinalState == FinalState.SUCCEEDED) { log.trace("Successfully produced messages to {} with base offset {}.", topicPartition, baseOffset); - finalState = FinalState.SUCCEEDED; } else { log.trace("Failed to produce messages to {}.", topicPartition, exception); - finalState = FinalState.FAILED; } - if (!this.finalState.compareAndSet(null, finalState)) { - if (this.finalState.get() == FinalState.ABORTED) { - log.debug("ProduceResponse returned for {} after batch had already been aborted.", topicPartition); - return false; - } else { - throw new IllegalStateException("Batch has already been completed in final state " + this.finalState.get()); + if (this.finalState.compareAndSet(null, tryFinalState)) { + completeFutureAndFireCallbacks(baseOffset, logAppendTime, exception); + return true; + } + + if (this.finalState.get() != FinalState.SUCCEEDED) { + if (tryFinalState == FinalState.SUCCEEDED) { + // Log if a previously unsuccessful batch succeeded later on. + log.debug("ProduceResponse returned {} for {} after batch had already been {}.", tryFinalState, + topicPartition, this.finalState.get()); } + // FAILED --> ABORTED and ABORTED --> FAILED transitions are ignored. + } else { + // A SUCCESSFUL batch must not attempt another state change. + throw new IllegalStateException("A " + this.finalState.get() + " batch must not attempt another state change to " + tryFinalState); } - completeFutureAndFireCallbacks(baseOffset, logAppendTime, exception); - return true; + return false; } private void completeFutureAndFireCallbacks(long baseOffset, long logAppendTime, RuntimeException exception) { @@ -300,29 +316,33 @@ public String toString() { } /** - * A batch whose metadata is not available should be expired if one of the following is true: - *
    - *
  1. the batch is not in retry AND request timeout has elapsed after it is ready (full or linger.ms has reached). - *
  2. the batch is in retry AND request timeout has elapsed after the backoff period ended. - *
+ * Expire the batch if current time has passed the delivery timeout of the batch. + * The delivery timeout is measured from batch creation time instead of batch close time + * because in some cases batch close may be arbitrarily delayed. * This methods closes this batch and sets {@code expiryErrorMessage} if the batch has timed out. */ - boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) { - if (!this.inRetry() && isFull && requestTimeoutMs < (now - this.lastAppendTime)) - expiryErrorMessage = (now - this.lastAppendTime) + " ms has passed since last append"; - else if (!this.inRetry() && requestTimeoutMs < (createdTimeMs(now) - lingerMs)) - expiryErrorMessage = (createdTimeMs(now) - lingerMs) + " ms has passed since batch creation plus linger time"; - else if (this.inRetry() && requestTimeoutMs < (waitedTimeMs(now) - retryBackoffMs)) - expiryErrorMessage = (waitedTimeMs(now) - retryBackoffMs) + " ms has passed since last attempt plus backoff time"; + boolean maybeExpire(long deliveryTimeoutMs, long now) { + if (deliveryTimeoutMs <= (now - this.createdMs)) { + expire(now); + return true; + } + return false; + } + + /** + * Expire the batch. This methods closes this batch and sets {@code expiryErrorMessage}. + */ + void expire(long now) { + expiryErrorMessage = (now - this.createdMs) + " ms has passed since batch creation"; + abortRecordAppends(); + } - boolean expired = expiryErrorMessage != null; - if (expired) - abortRecordAppends(); - return expired; + public FinalState finalState() { + return this.finalState.get(); } /** - * If {@link #maybeExpire(int, long, long, long, boolean)} returned true, the sender will fail the batch with + * If {@link #maybeExpire(long, long)} returned true, the sender will fail the batch with * the exception returned by this method. * @return An exception indicating the batch expired. */ @@ -468,4 +488,9 @@ public boolean sequenceHasBeenReset() { return reopened; } + boolean soonToExpire(long deliveryTimeoutMs, long now, int requestTimeoutMs) { + // We don't add anything in deliveryTimeoutMs because it may overflow + // when deliveryTimeoutMs==Long.MAX_VALUE + return deliveryTimeoutMs <= now + requestTimeoutMs - createdMs; + } } 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 ba8c28ece27f6..a4fd77e128865 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 @@ -16,6 +16,21 @@ */ package org.apache.kafka.clients.producer.internals; +import java.nio.ByteBuffer; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.common.Cluster; @@ -33,10 +48,10 @@ import org.apache.kafka.common.record.AbstractRecords; import org.apache.kafka.common.record.CompressionRatioEstimator; import org.apache.kafka.common.record.CompressionType; -import org.apache.kafka.common.record.Record; -import org.apache.kafka.common.record.RecordBatch; import org.apache.kafka.common.record.MemoryRecords; import org.apache.kafka.common.record.MemoryRecordsBuilder; +import org.apache.kafka.common.record.Record; +import org.apache.kafka.common.record.RecordBatch; import org.apache.kafka.common.record.TimestampType; import org.apache.kafka.common.utils.CopyOnWriteMap; import org.apache.kafka.common.utils.LogContext; @@ -44,20 +59,6 @@ import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; -import java.nio.ByteBuffer; -import java.util.ArrayDeque; -import java.util.ArrayList; -import java.util.Collections; -import java.util.Deque; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.atomic.AtomicInteger; - /** * This class acts as a queue that accumulates records into {@link MemoryRecords} * instances to be sent to the server. @@ -75,6 +76,8 @@ public final class RecordAccumulator { private final CompressionType compression; private final long lingerMs; private final long retryBackoffMs; + private final int requestTimeoutMs; + private final long deliveryTimeoutMs; private final BufferPool free; private final Time time; private final ApiVersions apiVersions; @@ -85,12 +88,15 @@ public final class RecordAccumulator { private int drainIndex; private final TransactionManager transactionManager; + // A per-partition queue of batches ordered by creation time for quick access of the oldest batch + private final ConcurrentMap> soonToExpireInFlightBatches; + private long nextBatchExpiryTimeMs; // the earliest time (absolute) a batch will expire. + /** * Create a new record accumulator * * @param logContext The log context used for logging * @param batchSize The size to use when allocating {@link MemoryRecords} instances - * @param totalSize The maximum memory the record accumulator can use. * @param compression The compression codec for the records * @param lingerMs An artificial delay time to add before declaring a records instance that isn't full ready for * sending. This allows time for more records to arrive. Setting a non-zero lingerMs will trade off some @@ -105,14 +111,17 @@ public final class RecordAccumulator { */ public RecordAccumulator(LogContext logContext, int batchSize, - long totalSize, CompressionType compression, long lingerMs, long retryBackoffMs, + int requestTimeoutMs, + long deliveryTimeoutMs, Metrics metrics, + String metricGrpName, Time time, ApiVersions apiVersions, - TransactionManager transactionManager) { + TransactionManager transactionManager, + BufferPool bufferPool) { this.log = logContext.logger(RecordAccumulator.class); this.drainIndex = 0; this.closed = false; @@ -122,14 +131,17 @@ public RecordAccumulator(LogContext logContext, this.compression = compression; this.lingerMs = lingerMs; this.retryBackoffMs = retryBackoffMs; + this.requestTimeoutMs = requestTimeoutMs; + this.deliveryTimeoutMs = deliveryTimeoutMs; + this.nextBatchExpiryTimeMs = Long.MAX_VALUE; this.batches = new CopyOnWriteMap<>(); - String metricGrpName = "producer-metrics"; - this.free = new BufferPool(totalSize, batchSize, metrics, time, metricGrpName); + this.free = bufferPool; this.incomplete = new IncompleteBatches(); this.muted = new HashSet<>(); this.time = time; this.apiVersions = apiVersions; this.transactionManager = transactionManager; + this.soonToExpireInFlightBatches = new ConcurrentHashMap<>(); registerMetrics(metrics, metricGrpName); } @@ -265,38 +277,63 @@ private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, H return null; } + private void maybeUpdateNextBatchExpiryTime(ProducerBatch batch) { + // We determine the earliest time any batch may expire. We use + // this value later on to determine the max time we can sleep in poll. + // We assume that the batch at the front of the deque will always be the next to expire. + // This may not be true if max.in.flight.requests.per.connection > 1 and retries happen. + // Watch for overflow in createdMs + deliveryTimeoutMs when deliveryTimeoutMs is Long.MAX_VALUE + nextBatchExpiryTimeMs = (batch.createdMs + deliveryTimeoutMs < 0) ? nextBatchExpiryTimeMs + : Math.min(nextBatchExpiryTimeMs, batch.createdMs + deliveryTimeoutMs); + } + /** * Get a list of batches which have been sitting in the accumulator too long and need to be expired. */ - public List expiredBatches(int requestTimeout, long now) { + public List expiredBatches(long now) { List expiredBatches = new ArrayList<>(); + for (Map.Entry> entry : this.batches.entrySet()) { - Deque dq = entry.getValue(); - TopicPartition tp = entry.getKey(); - // We only check if the batch should be expired if the partition does not have a batch in flight. - // This is to prevent later batches from being expired while an earlier batch is still in progress. - // Note that `muted` is only ever populated if `max.in.flight.request.per.connection=1` so this protection - // is only active in this case. Otherwise the expiration order is not guaranteed. - if (!muted.contains(tp)) { - synchronized (dq) { - // iterate over the batches and expire them if they have been in the accumulator for more than requestTimeOut - ProducerBatch lastBatch = dq.peekLast(); - Iterator batchIterator = dq.iterator(); - while (batchIterator.hasNext()) { - ProducerBatch batch = batchIterator.next(); - boolean isFull = batch != lastBatch || batch.isFull(); - // Check if the batch has expired. Expired batches are closed by maybeExpire, but callbacks - // are invoked after completing the iterations, since sends invoked from callbacks - // may append more batches to the deque being iterated. The batch is deallocated after - // callbacks are invoked. - if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) { + // Expire inflight batches in the order of draining. Expire if the final state of the + // batch is not known by (batch's creation time + deliveryTimeoutMs). + List mayExpireBatches = soonToExpireInFlightBatches.get(entry.getKey()); + if (mayExpireBatches != null) { + Iterator iter = mayExpireBatches.iterator(); + while (iter.hasNext()) { + ProducerBatch batch = iter.next(); + if(batch.maybeExpire(deliveryTimeoutMs, now)) { + iter.remove(); + if (batch.finalState() == null) { + batch.expire(now); expiredBatches.add(batch); - batchIterator.remove(); - } else { - // Stop at the first batch that has not expired. - break; } } + else { + maybeUpdateNextBatchExpiryTime(batch); + break; + } + } + } + + Deque dq = entry.getValue(); + synchronized (dq) { + // iterate over the batches and expire them if they have been in the accumulator for more than requestTimeOut + ProducerBatch lastBatch = dq.peekLast(); + Iterator batchIterator = dq.iterator(); + while (batchIterator.hasNext()) { + ProducerBatch batch = batchIterator.next(); + // Check if the batch has expired. Expired batches are closed by maybeExpire, but callbacks + // are invoked after completing the iterations, since sends invoked from callbacks + // may append more batches to the deque being iterated. The batch is deallocated after + // callbacks are invoked. + if (batch.maybeExpire(deliveryTimeoutMs, now)) { + expiredBatches.add(batch); + batchIterator.remove(); + } else { + maybeUpdateNextBatchExpiryTime(batch); + // Stop at the first batch that has not expired. + break; + } } } } @@ -304,10 +341,12 @@ public List expiredBatches(int requestTimeout, long now) { } /** - * Re-enqueue the given record batch in the accumulator to retry + * Re-enqueue the given record batch in the accumulator to retry. + * Remove the batch from the list of inflight batches. */ public void reenqueue(ProducerBatch batch, long now) { batch.reenqueued(now); + maybeRemoveFromSoonToExpire(batch); Deque deque = getOrCreateDeque(batch.topicPartition); synchronized (deque) { if (transactionManager != null) @@ -317,6 +356,13 @@ public void reenqueue(ProducerBatch batch, long now) { } } + private void maybeRemoveFromSoonToExpire(ProducerBatch batch) { + List soonToExpireBatches = soonToExpireInFlightBatches.get(batch.topicPartition); + if (soonToExpireBatches != null) { + soonToExpireBatches.remove(batch); + } + } + /** * Split the big batch that has been rejected and reenqueue the split batches in to the accumulator. * @return the number of split batches. @@ -572,6 +618,19 @@ public Map> drain(Cluster cluster, batch.close(); size += batch.records().sizeInBytes(); ready.add(batch); + // Put this batch in the list of soon-to-expire-inflight-batches because we + // might have to expire it if we don't know its final state by deliveryTimeoutMs + // If the expiry is farther than requestTimeoutMs, we don't have to keep + // track of this batch because it will either succeed or fail (due to + // request timeout) much sooner. + if (batch.soonToExpire(deliveryTimeoutMs, now, requestTimeoutMs)) { + List inflightBatchList = soonToExpireInFlightBatches.get(batch.topicPartition); + if (inflightBatchList == null) { + inflightBatchList = new LinkedList<>(); + soonToExpireInFlightBatches.put(tp, inflightBatchList); + } + inflightBatchList.add(batch); + } batch.drained(now); } } @@ -583,9 +642,39 @@ public Map> drain(Cluster cluster, } while (start != drainIndex); batches.put(node.id(), ready); } + + maybeUpdateNextExpiryTime(now); return batches; } + /** + * The earliest absolute time a batch will expire (in milliseconds) + */ + public Long nextExpiryTimeMs() { + return this.nextBatchExpiryTimeMs; + } + + /** + * If there are batches to soon to expire, but now is past the last known earliest expiry time + * we reset the expiry time to now, because it's the smallest possible valid absolute time in future. + * This could happen when there were no ready nodes for a while and clock went past the last known earliest + * expiry time. If there're no batches to soon expire, we simply reset it to the largest possible value. + * This ensures that next delivery timeout keeps moving ahead. + */ + void maybeUpdateNextExpiryTime(long now) { + if (now >= nextBatchExpiryTimeMs) { + if (soonToExpireInFlightBatches.isEmpty()) { + nextBatchExpiryTimeMs = Long.MAX_VALUE; + } else { + nextBatchExpiryTimeMs = now; + } + } + } + + ConcurrentMap> soonToExpireInFlightBatches() { + return soonToExpireInFlightBatches; + } + private Deque getDeque(TopicPartition tp) { return batches.get(tp); } 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 426b273b88538..5a9cb30090ef4 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 @@ -277,7 +277,7 @@ private long sendProducerData(long now) { } } - List expiredBatches = this.accumulator.expiredBatches(this.requestTimeout, now); + List expiredBatches = this.accumulator.expiredBatches(now); // 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. @@ -298,6 +298,7 @@ private long sendProducerData(long now) { // that isn't yet sendable (e.g. lingering, backing off). Note that this specifically does not include nodes // with sendable data that aren't ready to send since they would cause busy looping. long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout); + pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now); if (!result.readyNodes.isEmpty()) { log.trace("Nodes with data ready to send: {}", result.readyNodes); // if some partitions are already ready to be sent, the select time would be 0; @@ -635,6 +636,7 @@ private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, */ private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response) { return batch.attempts() < this.retries && + batch.finalState() == null && ((response.error.exception() instanceof RetriableException) || (transactionManager != null && transactionManager.canRetry(response, batch))); } diff --git a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java index c9efb82e061c5..7ad4c31fb5057 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java +++ b/clients/src/main/java/org/apache/kafka/common/config/ConfigDef.java @@ -57,7 +57,7 @@ * Map<String, String> props = new HashMap<>(); * props.put("config_with_default", "some value"); * props.put("config_with_dependents", "some other value"); - * + * * Map<String, Object> configs = defs.parse(props); * // will return "some value" * String someConfig = (String) configs.get("config_with_default"); @@ -595,10 +595,10 @@ private void validate(String name, Map parsed, Map recommendedValues = key.recommender.validValues(name, parsed); @@ -845,9 +845,9 @@ public static class Range implements Validator { private final Number min; private final Number max; - private Range(Number min, Number max) { + private Range(Number min, Number maxInclusive) { this.min = min; - this.max = max; + this.max = maxInclusive; } /** @@ -860,7 +860,7 @@ public static Range atLeast(Number min) { } /** - * A numeric range that checks both the upper and lower bound + * A numeric range that checks both the upper (inclusive) and lower bound */ public static Range between(Number min, Number max) { return new Range(min, max); diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java index 2f89d7949e702..07dbfc863a605 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java @@ -226,40 +226,30 @@ public void testSplitPreservesMagicAndCompressionType() { } /** - * A {@link ProducerBatch} configured using a very large linger value and a timestamp preceding its create - * time is interpreted correctly as not expired when the linger time is larger than the difference - * between now and create time by {@link ProducerBatch#maybeExpire(int, long, long, long, boolean)}. + * A {@link ProducerBatch} configured using a timestamp preceding its create time is interpreted correctly + * as not expired by {@link ProducerBatch#maybeExpire(long, long)}. */ @Test - public void testLargeLingerOldNowExpire() { + public void testBatchExpiration() { + long deliveryTimeoutMs = 10240; ProducerBatch batch = new ProducerBatch(new TopicPartition("topic", 1), memoryRecordsBuilder, now); // Set `now` to 2ms before the create time. - assertFalse(batch.maybeExpire(10240, 100L, now - 2L, Long.MAX_VALUE, false)); + assertFalse(batch.maybeExpire(deliveryTimeoutMs, now - 2)); + // Set `now` to deliveryTimeoutMs. + assertTrue(batch.maybeExpire(deliveryTimeoutMs, now + deliveryTimeoutMs)); } /** - * A {@link ProducerBatch} configured using a very large retryBackoff value with retry = true and a timestamp - * preceding its create time is interpreted correctly as not expired when the retryBackoff time is larger than the - * difference between now and create time by {@link ProducerBatch#maybeExpire(int, long, long, long, boolean)}. + * A {@link ProducerBatch} configured using a timestamp preceding its create time is interpreted correctly + * * as not expired by {@link ProducerBatch#maybeExpire(long, long)}. */ @Test - public void testLargeRetryBackoffOldNowExpire() { + public void testBatchExpirationAfterReenqueue() { ProducerBatch batch = new ProducerBatch(new TopicPartition("topic", 1), memoryRecordsBuilder, now); // Set batch.retry = true batch.reenqueued(now); // Set `now` to 2ms before the create time. - assertFalse(batch.maybeExpire(10240, Long.MAX_VALUE, now - 2L, 10240L, false)); - } - - /** - * A {@link ProducerBatch#maybeExpire(int, long, long, long, boolean)} call with a now value before the create - * time of the ProducerBatch is correctly recognized as not expired when invoked with parameter isFull = true. - */ - @Test - public void testLargeFullOldNowExpire() { - ProducerBatch batch = new ProducerBatch(new TopicPartition("topic", 1), memoryRecordsBuilder, now); - // Set `now` to 2ms before the create time. - assertFalse(batch.maybeExpire(10240, 10240L, now - 2L, 10240L, true)); + assertFalse(batch.maybeExpire(10240, now - 2L)); } @Test diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java index d486c10a94e16..4ee2de227c749 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java @@ -332,10 +332,17 @@ public void testNextReadyCheckDelay() throws Exception { @Test public void testRetryBackoff() throws Exception { - long lingerMs = Long.MAX_VALUE / 4; - long retryBackoffMs = Long.MAX_VALUE / 2; - final RecordAccumulator accum = new RecordAccumulator(logContext, 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024, - CompressionType.NONE, lingerMs, retryBackoffMs, metrics, time, new ApiVersions(), null); + long lingerMs = Integer.MAX_VALUE / 16; + long retryBackoffMs = Integer.MAX_VALUE / 8; + int requestTimeoutMs = Integer.MAX_VALUE / 4; + long deliveryTimeoutMs = Integer.MAX_VALUE; + long totalSize = 10 * 1024; + int batchSize = 1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD; + String metricGrpName = "producer-metrics"; + + final RecordAccumulator accum = new RecordAccumulator(logContext, batchSize, + CompressionType.NONE, lingerMs, retryBackoffMs, requestTimeoutMs, deliveryTimeoutMs, metrics, metricGrpName, time, new ApiVersions(), null, + new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)); long now = time.milliseconds(); accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); @@ -509,77 +516,51 @@ public void onCompletion(RecordMetadata metadata, Exception exception) { assertTrue(accum.hasIncomplete()); } - @Test - public void testExpiredBatches() throws InterruptedException { - long retryBackoffMs = 100L; + private void doExpireBatchSingle(long deliveryTimeoutMs) throws InterruptedException { long lingerMs = 3000L; - int requestTimeout = 60; + List muteStates = Arrays.asList(false, true); + Set readyNodes = null; + List expiredBatches = null; // test case assumes that the records do not fill the batch completely int batchSize = 1025; - RecordAccumulator accum = createTestRecordAccumulator( - batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, CompressionType.NONE, lingerMs); - int appends = expectedNumAppends(batchSize); + RecordAccumulator accum = createTestRecordAccumulator(deliveryTimeoutMs, + batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, CompressionType.NONE, lingerMs); - // Test batches not in retry - for (int i = 0; i < appends; i++) { + // Make the batches ready due to linger. + // These batches are not in retry + for (Boolean mute: muteStates) { accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); + assertEquals("No partition should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); + + time.sleep(lingerMs); + readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; + assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); + expiredBatches = accum.expiredBatches(time.milliseconds()); + assertEquals("The batch should not expire when just linger has passed", 0, expiredBatches.size()); + + if (mute) + accum.mutePartition(tp1); + else + accum.unmutePartition(tp1); + + // Advance the clock to expire the batch. + time.sleep(deliveryTimeoutMs - lingerMs); + expiredBatches = accum.expiredBatches(time.milliseconds()); + assertEquals("The batch may expire when the partition is muted", 1, expiredBatches.size()); assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); } - // Make the batches ready due to batch full - accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); - Set readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; - assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); - // Advance the clock to expire the batch. - time.sleep(requestTimeout + 1); - accum.mutePartition(tp1); - List expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should not be expired when the partition is muted", 0, expiredBatches.size()); - - accum.unmutePartition(tp1); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should be expired", 1, expiredBatches.size()); - assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); - - // Advance the clock to make the next batch ready due to linger.ms - time.sleep(lingerMs); - assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); - time.sleep(requestTimeout + 1); - - accum.mutePartition(tp1); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should not be expired when metadata is still available and partition is muted", 0, expiredBatches.size()); - - accum.unmutePartition(tp1); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should be expired when the partition is not muted", 1, expiredBatches.size()); - assertEquals("No partitions should be ready.", 0, accum.ready(cluster, time.milliseconds()).readyNodes.size()); - - // Test batches in retry. - // Create a retried batch - accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); - time.sleep(lingerMs); - readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; - assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); - Map> drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); - assertEquals("There should be only one batch.", drained.get(node1.id()).size(), 1); - time.sleep(1000L); - accum.reenqueue(drained.get(node1.id()).get(0), time.milliseconds()); - - // test expiration. - time.sleep(requestTimeout + retryBackoffMs); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should not be expired.", 0, expiredBatches.size()); - time.sleep(1L); + } - accum.mutePartition(tp1); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should not be expired when the partition is muted", 0, expiredBatches.size()); + @Test + public void testExpiredBatchSingle() throws InterruptedException { + doExpireBatchSingle(3200L); + } - accum.unmutePartition(tp1); - expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds()); - assertEquals("The batch should be expired when the partition is not muted.", 1, expiredBatches.size()); + @Test + public void testExpiredBatchSingleMaxValue() throws InterruptedException { + doExpireBatchSingle(Long.MAX_VALUE); } @Test @@ -623,10 +604,18 @@ public void testIdempotenceWithOldMagic() throws InterruptedException { // Simulate talking to an older broker, ie. one which supports a lower magic. ApiVersions apiVersions = new ApiVersions(); int batchSize = 1025; + int requestTimeoutMs = 1600; + long deliveryTimeoutMs = 3200L; + long lingerMs = 10L; + long retryBackoffMs = 100L; + long totalSize = 10 * batchSize; + String metricGrpName = "producer-metrics"; + apiVersions.update("foobar", NodeApiVersions.create(Arrays.asList(new ApiVersionsResponse.ApiVersion(ApiKeys.PRODUCE.id, (short) 0, (short) 2)))); - RecordAccumulator accum = new RecordAccumulator(logContext, batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, - CompressionType.NONE, 10, 100L, metrics, time, apiVersions, new TransactionManager()); + RecordAccumulator accum = new RecordAccumulator(logContext, batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, + CompressionType.NONE, lingerMs, retryBackoffMs, requestTimeoutMs, deliveryTimeoutMs, metrics, metricGrpName, time, apiVersions, new TransactionManager(), + new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)); accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); } @@ -742,6 +731,77 @@ public void testSplitFrequency() throws InterruptedException { } } + @Test + public void testSoonToExpireBatchesArePickedUpForExpiry() throws InterruptedException { + long lingerMs = 500L; + int batchSize = 1025; + + RecordAccumulator accum = createTestRecordAccumulator( + batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, CompressionType.NONE, lingerMs); + + accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); + Set readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; + Map> drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); + assertTrue(drained.isEmpty()); + assertTrue(accum.soonToExpireInFlightBatches().isEmpty()); + + // advanced clock and send one batch out but it should not be included in soon to expire inflight + // batches because batch's expiry is quite far. + time.sleep(lingerMs + 1); + readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; + drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); + assertEquals("A batch did not drain after linger", 1, drained.size()); + assertTrue(accum.soonToExpireInFlightBatches().isEmpty()); + + // Queue another batch and advance clock such that batch expiry time is earlier than request timeout. + accum.append(tp2, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs); + time.sleep(lingerMs * 4); + + // Now drain and check that accumulator picked up the drained batch because its expiry is soon. + readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; + drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); + assertEquals("A batch did not drain after linger", 1, drained.size()); + assertEquals("A soon to expire batch was ignored", 1, accum.soonToExpireInFlightBatches().size()); + } + + @Test + public void testExpiredBatchesRetry() throws InterruptedException { + long lingerMs = 3000L; + long rtt = 1000; + long deliveryTimeoutMs = 3200L; + Set readyNodes = null; + List expiredBatches = null; + List muteStates = Arrays.asList(false, true); + + // test case assumes that the records do not fill the batch completely + int batchSize = 1025; + + RecordAccumulator accum = createTestRecordAccumulator( + batchSize + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * batchSize, CompressionType.NONE, lingerMs); + + // Test batches in retry. + for (Boolean mute: muteStates) { + accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, 0); + time.sleep(lingerMs); + readyNodes = accum.ready(cluster, time.milliseconds()).readyNodes; + assertEquals("Our partition's leader should be ready", Collections.singleton(node1), readyNodes); + Map> drained = accum.drain(cluster, readyNodes, Integer.MAX_VALUE, time.milliseconds()); + assertEquals("There should be only one batch.", 1, drained.get(node1.id()).size()); + time.sleep(rtt); + accum.reenqueue(drained.get(node1.id()).get(0), time.milliseconds()); + + if (mute) + accum.mutePartition(tp1); + else + accum.unmutePartition(tp1); + + // test expiration + time.sleep(deliveryTimeoutMs - rtt); + expiredBatches = accum.expiredBatches(time.milliseconds()); + assertEquals("The batch in retry may expire.", 1, expiredBatches.size()); + } + } + private int prepareSplitBatches(RecordAccumulator accum, long seed, int recordSize, int numRecords) throws InterruptedException { Random random = new Random(); @@ -837,20 +897,34 @@ private int expectedNumAppends(int batchSize) { } } + + private RecordAccumulator createTestRecordAccumulator(int batchSize, long totalSize, CompressionType type, long lingerMs) { + long deliveryTimeoutMs = 3200L; + return createTestRecordAccumulator(deliveryTimeoutMs, batchSize, totalSize, type, lingerMs); + } + /** * Return a test RecordAccumulator instance */ - private RecordAccumulator createTestRecordAccumulator(int batchSize, long totalSize, CompressionType type, long lingerMs) { + private RecordAccumulator createTestRecordAccumulator(long deliveryTimeoutMs, int batchSize, long totalSize, CompressionType type, long lingerMs) { + long retryBackoffMs = 100L; + int requestTimeoutMs = 1600; + String metricGrpName = "producer-metrics"; + return new RecordAccumulator( - logContext, - batchSize, - totalSize, - type, - lingerMs, - 100L, - metrics, - time, - new ApiVersions(), - null); + logContext, + batchSize, + type, + lingerMs, + retryBackoffMs, + requestTimeoutMs, + deliveryTimeoutMs, + metrics, + metricGrpName, + time, + new ApiVersions(), + null, + new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)); } } + 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 1ce8e5a4bbdb4..a45e04650ed9c 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 @@ -16,6 +16,21 @@ */ package org.apache.kafka.clients.producer.internals; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.IdentityHashMap; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.ApiVersions; import org.apache.kafka.clients.ClientRequest; import org.apache.kafka.clients.Metadata; @@ -62,6 +77,7 @@ import org.apache.kafka.common.requests.ResponseHeader; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.common.utils.Time; import org.apache.kafka.test.DelayedReceive; import org.apache.kafka.test.MockSelector; import org.apache.kafka.test.TestUtils; @@ -69,25 +85,10 @@ import org.junit.Before; import org.junit.Test; -import java.nio.ByteBuffer; -import java.util.Collections; -import java.util.Deque; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; - import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class SenderTest { @@ -969,6 +970,9 @@ public void testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatch // 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 + // We separate the two appends by 1 second so that the two batches + // don't expire at the same time. + time.sleep(1000L); Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; sender.run(time.milliseconds()); // send request @@ -979,7 +983,9 @@ public void testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatch sender.run(time.milliseconds()); // receive first response Node node = this.cluster.nodes().get(0); - time.sleep(10000L); + // We add 600 millis to expire the first batch but not the second. + // Note deliveryTimeoutMs is 1500. + time.sleep(600L); client.disconnect(node.idString()); client.blackout(node, 10); @@ -1027,6 +1033,9 @@ public void testExpiryOfFirstBatchShouldCauseResetIfFutureBatchesFail() throws E // 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 + // We separate the two appends by 1 second so that the two batches + // don't expire at the same time. + time.sleep(1000L); Future request2 = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; sender.run(time.milliseconds()); // send request @@ -1037,7 +1046,9 @@ public void testExpiryOfFirstBatchShouldCauseResetIfFutureBatchesFail() throws E sender.run(time.milliseconds()); // receive first response Node node = this.cluster.nodes().get(0); - time.sleep(10000L); + // We add 600 millis to expire the first batch but not the second. + // Note deliveryTimeoutMs is 1500. + time.sleep(600L); client.disconnect(node.idString()); client.blackout(node, 10); @@ -1656,7 +1667,7 @@ public void testSequenceNumberIncrement() throws InterruptedException { int maxRetries = 10; Metrics m = new Metrics(); SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m); - + Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); @@ -1740,7 +1751,7 @@ public void testResetWhenOutOfOrderSequenceReceived() throws InterruptedExceptio int maxRetries = 10; Metrics m = new Metrics(); SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m); - + Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); @@ -1789,11 +1800,16 @@ private void testSplitBatchAndSend(TransactionManager txnManager, TopicPartition tp) throws Exception { int maxRetries = 1; String topic = tp.topic(); + int requestTimeoutMs = 1500; + long deliveryTimeoutMs = 3000L; + long totalSize = 1024 * 1024; + String metricGrpName = "producer-metrics"; // Set a good compression ratio. CompressionRatioEstimator.setEstimation(topic, CompressionType.GZIP, 0.2f); try (Metrics m = new Metrics()) { - accumulator = new RecordAccumulator(logContext, batchSize, 1024 * 1024, CompressionType.GZIP, 0L, 0L, m, time, - new ApiVersions(), txnManager); + accumulator = new RecordAccumulator(logContext, batchSize, CompressionType.GZIP, + 0L, 0L, requestTimeoutMs, deliveryTimeoutMs, m, metricGrpName, time, new ApiVersions(), txnManager, + new BufferPool(totalSize, batchSize, metrics, time, "producer-internal-metrics")); SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m); Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 1000L, txnManager, new ApiVersions()); @@ -1866,6 +1882,145 @@ private void testSplitBatchAndSend(TransactionManager txnManager, } } + @Test + public void testNoDoubleDeallocation() throws Exception { + long deliverTimeoutMs = 1500L; + long totalSize = 1024 * 1024; + String metricGrpName = "producer-custom-metrics"; + MatchingBufferPool pool = new MatchingBufferPool(totalSize, batchSize, metrics, time, metricGrpName); + setupWithTransactionState(null, false, pool); + + // 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 + assertEquals(1, client.inFlightRequestCount()); + time.sleep(deliverTimeoutMs); + sender.run(time.milliseconds()); // expire the batch + assertTrue(request1.isDone()); + // make sure the batch got deallocated right after expiry + assertTrue(pool.allMatch()); + + assertEquals(1, client.inFlightRequestCount()); + client.respond(produceResponse(tp0, -1, Errors.NOT_LEADER_FOR_PARTITION, -1)); // return a retriable error + + sender.run(time.milliseconds()); // receive first response and do not reenqueue. + assertEquals(0, client.inFlightRequestCount()); + sender.run(time.milliseconds()); // run again and must not send anything. + assertEquals(0, client.inFlightRequestCount()); + assertTrue(pool.allMatch()); + } + + @Test + public void testInflightBatchesExpireOnDeliveryTimeout() throws InterruptedException { + long deliveryTimeoutMs = 1500L; + + setupWithTransactionState(null, true, null); + + // Send first ProduceRequest + Future request = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future; + sender.run(time.milliseconds()); // send request + assertEquals(1, client.inFlightRequestCount()); + + Map responseMap = new HashMap<>(); + responseMap.put(tp0, new ProduceResponse.PartitionResponse(Errors.NONE, 0L, 0L, 0L)); + client.respond(new ProduceResponse(responseMap)); + + time.sleep(deliveryTimeoutMs); + sender.run(time.milliseconds()); // receive first response + try { + request.get(); + fail("The expired batch should throw a TimeoutException"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof TimeoutException); + } + } + + @Test + public void testWhenFirstBatchExpireNoSendSecondBatchIfGuaranteeOrder() throws InterruptedException { + long deliveryTimeoutMs = 1500L; + + setupWithTransactionState(null, true, null); + + // Send first ProduceRequest + accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT); + sender.run(time.milliseconds()); // send request + assertEquals(1, client.inFlightRequestCount()); + + time.sleep(deliveryTimeoutMs / 2); + + // Send second ProduceRequest + accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT); + sender.run(time.milliseconds()); // must not send request because the partition is muted + assertEquals(1, client.inFlightRequestCount()); + + time.sleep(deliveryTimeoutMs / 2); // expire the first batch only + sender.run(time.milliseconds()); + assertEquals(1, client.inFlightRequestCount()); + + client.respond(produceResponse(tp0, 0L, Errors.NONE, 0, 0L)); + sender.run(time.milliseconds()); // receive response (offset=0) + assertEquals(0, client.inFlightRequestCount()); + + sender.run(time.milliseconds()); // Drain the second request only this time + assertEquals(1, client.inFlightRequestCount()); + } + + @Test + public void testExpiredBatchDoesNotRetry() throws Exception { + long deliverTimeoutMs = 1500L; + setupWithTransactionState(null, false, null); + + // 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 + assertEquals(1, client.inFlightRequestCount()); + time.sleep(deliverTimeoutMs); + sender.run(time.milliseconds()); // expire the batch + assertTrue(request1.isDone()); + assertEquals(1, client.inFlightRequestCount()); + + Map responseMap = new HashMap<>(); + responseMap.put(tp0, new ProduceResponse.PartitionResponse(Errors.NONE, 0L, 0L, 0L)); + client.respond(produceResponse(tp0, -1, Errors.NOT_LEADER_FOR_PARTITION, -1)); // return a retriable error + + sender.run(time.milliseconds()); // receive first response and do not reenqueue. + assertEquals(0, client.inFlightRequestCount()); + sender.run(time.milliseconds()); // run again and must not send anything. + assertEquals(0, client.inFlightRequestCount()); + } + + private class MatchingBufferPool extends BufferPool { + IdentityHashMap allocatedBuffers; + + MatchingBufferPool(long totalSize, int batchSize, Metrics metrics, Time time, String metricGrpName) { + super(totalSize, batchSize, metrics, time, metricGrpName); + allocatedBuffers = new IdentityHashMap<>(); + } + + @Override + public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException { + ByteBuffer buffer = super.allocate(size, maxTimeToBlockMs); + allocatedBuffers.put(buffer, Boolean.TRUE); + return buffer; + } + + @Override + public void deallocate(ByteBuffer buffer, int size) { + if (!allocatedBuffers.containsKey(buffer)) { + throw new IllegalStateException("Deallocating a buffer that is not allocated"); + } + allocatedBuffers.remove(buffer); + super.deallocate(buffer, size); + } + + public boolean allMatch() { + return allocatedBuffers.isEmpty(); + } + } + private MockClient.RequestMatcher produceRequestMatcher(final TopicPartition tp, final ProducerIdAndEpoch producerIdAndEpoch, final int sequence, @@ -1926,16 +2081,29 @@ private ProduceResponse produceResponse(TopicPartition tp, long offset, Errors e } private void setupWithTransactionState(TransactionManager transactionManager) { + setupWithTransactionState(transactionManager, false, null); + } + + private void setupWithTransactionState(TransactionManager transactionManager, boolean guaranteeOrder, BufferPool customPool) { + long totalSize = 1024 * 1024; + String metricGrpName = "producer-metrics"; Map metricTags = new LinkedHashMap<>(); metricTags.put("client-id", CLIENT_ID); MetricConfig metricConfig = new MetricConfig().tags(metricTags); 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(this.metrics); + BufferPool pool = (customPool == null) ? new BufferPool(totalSize, batchSize, metrics, time, metricGrpName) : customPool; + setupWithTransactionState(transactionManager, guaranteeOrder, metricTags, pool); + } - this.sender = new Sender(logContext, this.client, this.metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL, - Integer.MAX_VALUE, this.senderMetricsRegistry, this.time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); + private void setupWithTransactionState(TransactionManager transactionManager, boolean guaranteeOrder, Map metricTags, BufferPool pool) { + long deliveryTimeoutMs = 1500L; + int requestTimeoutMs = 1500; + String metricGrpName = "producer-metrics"; + this.accumulator = new RecordAccumulator(logContext, batchSize, CompressionType.NONE, 0L, 0L, + requestTimeoutMs, deliveryTimeoutMs, metrics, metricGrpName, time, apiVersions, transactionManager, pool); + this.senderMetricsRegistry = new SenderMetricsRegistry(this.metrics); + this.sender = new Sender(logContext, this.client, this.metadata, this.accumulator, guaranteeOrder, MAX_REQUEST_SIZE, ACKS_ALL, + Integer.MAX_VALUE, 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 6fcf48059672a..48e7a788bcb3f 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 @@ -118,6 +118,10 @@ public void setup() { Map metricTags = new LinkedHashMap<>(); metricTags.put("client-id", CLIENT_ID); int batchSize = 16 * 1024; + int requestTimeoutMs = 1500; + long deliveryTimeoutMs = 3000L; + long totalSize = 1024 * 1024; + String metricGrpName = "producer-metrics"; MetricConfig metricConfig = new MetricConfig().tags(metricTags); this.brokerNode = new Node(0, "localhost", 2211); this.transactionManager = new TransactionManager(logContext, transactionalId, transactionTimeoutMs, @@ -125,7 +129,7 @@ public void setup() { Metrics metrics = new Metrics(metricConfig, time); SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(metrics); - this.accumulator = new RecordAccumulator(logContext, batchSize, 1024 * 1024, CompressionType.NONE, 0L, 0L, metrics, time, apiVersions, transactionManager); + this.accumulator = new RecordAccumulator(logContext, batchSize, CompressionType.NONE, 0L, 0L, requestTimeoutMs, deliveryTimeoutMs, metrics, metricGrpName, time, apiVersions, transactionManager, new BufferPool(totalSize, batchSize, metrics, time, metricGrpName)); this.sender = new Sender(logContext, this.client, this.metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, MAX_RETRIES, senderMetrics, this.time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions); this.metadata.update(this.cluster, Collections.emptySet(), time.milliseconds()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index 1c6465855ff7b..b66f67cbdae69 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -128,6 +128,7 @@ public Worker( producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE)); producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1"); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE)); // User-specified overrides producerProps.putAll(config.originalsWithPrefix("producer.")); } diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index 4b8740600ff7c..af428898b581c 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -571,6 +571,10 @@ object TestUtils extends Logging { producerProps.put(ProducerConfig.RETRIES_CONFIG, retries.toString) producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs.toString) + // In case of overflow set maximum possible value for deliveryTimeoutMs + val deliveryTimeoutMs = if (lingerMs + requestTimeoutMs < 0) Long.MaxValue else lingerMs + requestTimeoutMs + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs.toString) + /* Only use these if not already set */ val defaultProps = Map( ProducerConfig.RETRY_BACKOFF_MS_CONFIG -> "100",