You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jg...@apache.org on 2017/06/07 20:02:06 UTC
kafka git commit: KAFKA-5385;
ProducerBatch expiry should go through Sender.failBatch
Repository: kafka
Updated Branches:
refs/heads/trunk effda15f1 -> 79db393ff
KAFKA-5385; ProducerBatch expiry should go through Sender.failBatch
Before this patch, we would call `producerBatch.done` directly from the accumulator when expiring batches. This meant that we would not transition to the `ABORTABLE_ERROR` state in the transaction manager, allowing other transactional requests (including Commits!) to go through, even though the produce failed.
This patch modifies the logic so that we call `Sender.failBatch` on every expired batch, thus ensuring that the transaction state is accurate.
Author: Apurva Mehta <ap...@confluent.io>
Reviewers: Guozhang Wang <wa...@gmail.com>, Jason Gustafson <ja...@confluent.io>
Closes #3252 from apurvam/KAFKA-5385-fail-transaction-if-batches-expire
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/79db393f
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/79db393f
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/79db393f
Branch: refs/heads/trunk
Commit: 79db393ffaca25fdaac0fef69d6fb082f1628649
Parents: effda15
Author: Apurva Mehta <ap...@confluent.io>
Authored: Wed Jun 7 13:01:54 2017 -0700
Committer: Jason Gustafson <ja...@confluent.io>
Committed: Wed Jun 7 13:01:54 2017 -0700
----------------------------------------------------------------------
.../producer/internals/ProducerBatch.java | 14 +-
.../producer/internals/RecordAccumulator.java | 15 +-
.../clients/producer/internals/Sender.java | 16 ++-
.../producer/internals/TransactionManager.java | 1 -
.../internals/RecordAccumulatorTest.java | 59 +-------
.../clients/producer/internals/SenderTest.java | 50 +++++++
.../internals/TransactionManagerTest.java | 138 +++++++++++++++++++
.../kafka/tools/TransactionalMessageCopier.java | 40 ++++--
8 files changed, 242 insertions(+), 91 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
----------------------------------------------------------------------
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 c7253a9..7f3ba15 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
@@ -264,7 +264,6 @@ public final class ProducerBatch {
* <li> the batch is in retry AND request timeout has elapsed after the backoff period ended.
* </ol>
* This methods closes this batch and sets {@code expiryErrorMessage} if the batch has timed out.
- * {@link #expirationDone()} must be invoked to complete the produce future and invoke callbacks.
*/
boolean maybeExpire(int requestTimeoutMs, long retryBackoffMs, long now, long lingerMs, boolean isFull) {
if (!this.inRetry() && isFull && requestTimeoutMs < (now - this.lastAppendTime))
@@ -276,20 +275,19 @@ public final class ProducerBatch {
boolean expired = expiryErrorMessage != null;
if (expired)
- close();
+ abort();
return expired;
}
/**
- * Completes the produce future with timeout exception and invokes callbacks.
- * This method should be invoked only if {@link #maybeExpire(int, long, long, long, boolean)}
- * returned true.
+ * If {@link #maybeExpire(int, long, long, long, boolean)} returned true, the sender will fail the batch with
+ * the exception returned by this method.
+ * @return An exception indicating the batch expired.
*/
- void expirationDone() {
+ TimeoutException timeoutException() {
if (expiryErrorMessage == null)
throw new IllegalStateException("Batch has not expired");
- this.done(-1L, NO_TIMESTAMP,
- new TimeoutException("Expiring " + recordCount + " record(s) for " + topicPartition + ": " + expiryErrorMessage));
+ return new TimeoutException("Expiring " + recordCount + " record(s) for " + topicPartition + ": " + expiryErrorMessage);
}
int attempts() {
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java
----------------------------------------------------------------------
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 5d4af74..7237b6d 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
@@ -263,12 +263,10 @@ public final class RecordAccumulator {
}
/**
- * Abort the batches that have been sitting in RecordAccumulator for more than the configured requestTimeout
- * due to metadata being unavailable
+ * Get a list of batches which have been sitting in the accumulator too long and need to be expired.
*/
- public List<ProducerBatch> abortExpiredBatches(int requestTimeout, long now) {
+ public List<ProducerBatch> expiredBatches(int requestTimeout, long now) {
List<ProducerBatch> expiredBatches = new ArrayList<>();
- int count = 0;
for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {
Deque<ProducerBatch> dq = entry.getValue();
TopicPartition tp = entry.getKey();
@@ -290,7 +288,6 @@ public final class RecordAccumulator {
// callbacks are invoked.
if (batch.maybeExpire(requestTimeout, retryBackoffMs, now, this.lingerMs, isFull)) {
expiredBatches.add(batch);
- count++;
batchIterator.remove();
} else {
// Stop at the first batch that has not expired.
@@ -300,14 +297,6 @@ public final class RecordAccumulator {
}
}
}
- if (!expiredBatches.isEmpty()) {
- log.trace("Expired {} batches in accumulator", count);
- for (ProducerBatch batch : expiredBatches) {
- batch.expirationDone();
- deallocate(batch);
- }
- }
-
return expiredBatches;
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
----------------------------------------------------------------------
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 1a5b321..b8d4ab9 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
@@ -66,6 +66,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP;
+
/**
* The background thread that handles the sending of produce requests to the Kafka cluster. This thread makes metadata
* requests to renew its view of the cluster and then sends produce requests to the appropriate nodes.
@@ -257,13 +259,16 @@ public class Sender implements Runnable {
}
}
- List<ProducerBatch> expiredBatches = this.accumulator.abortExpiredBatches(this.requestTimeout, now);
+ List<ProducerBatch> 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());
if (transactionManager != null && expiredBatch.inRetry()) {
needsTransactionStateReset = true;
}
@@ -573,13 +578,17 @@ public class Sender implements Runnable {
}
private void failBatch(ProducerBatch batch, ProduceResponse.PartitionResponse response, RuntimeException exception) {
+ failBatch(batch, response.baseOffset, response.logAppendTime, exception);
+ }
+
+ private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception) {
if (transactionManager != null) {
if (exception instanceof OutOfOrderSequenceException
&& !transactionManager.isTransactional()
&& transactionManager.hasProducerId(batch.producerId())) {
log.error("The broker received an out of order sequence number for topic-partition " +
"{} at offset {}. This indicates data loss on the broker, and should be investigated.",
- batch.topicPartition, response.baseOffset);
+ batch.topicPartition, baseOffset);
// Reset the transaction state since we have hit an irrecoverable exception and cannot make any guarantees
// about the previously committed message. Note that this will discard the producer id and sequence
@@ -593,8 +602,7 @@ public class Sender implements Runnable {
transactionManager.transitionToAbortableError(exception);
}
}
-
- batch.done(response.baseOffset, response.logAppendTime, exception);
+ batch.done(baseOffset, logAppendTime, exception);
this.accumulator.deallocate(batch);
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
----------------------------------------------------------------------
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 d959d7d..a761b31 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
@@ -1005,5 +1005,4 @@ public class TransactionManager {
reenqueue();
}
}
-
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/test/java/org/apache/kafka/clients/producer/internals/RecordAccumulatorTest.java
----------------------------------------------------------------------
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 7b9f26b..3929156 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
@@ -24,7 +24,6 @@ import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -57,13 +56,11 @@ 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 java.util.Arrays.asList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -421,11 +418,11 @@ public class RecordAccumulatorTest {
// Advance the clock to expire the batch.
time.sleep(requestTimeout + 1);
accum.mutePartition(tp1);
- List<ProducerBatch> expiredBatches = accum.abortExpiredBatches(requestTimeout, time.milliseconds());
+ List<ProducerBatch> 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.abortExpiredBatches(requestTimeout, time.milliseconds());
+ 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());
@@ -435,11 +432,11 @@ public class RecordAccumulatorTest {
time.sleep(requestTimeout + 1);
accum.mutePartition(tp1);
- expiredBatches = accum.abortExpiredBatches(requestTimeout, time.milliseconds());
+ 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.abortExpiredBatches(requestTimeout, time.milliseconds());
+ 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());
@@ -456,62 +453,20 @@ public class RecordAccumulatorTest {
// test expiration.
time.sleep(requestTimeout + retryBackoffMs);
- expiredBatches = accum.abortExpiredBatches(requestTimeout, time.milliseconds());
+ expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds());
assertEquals("The batch should not be expired.", 0, expiredBatches.size());
time.sleep(1L);
accum.mutePartition(tp1);
- expiredBatches = accum.abortExpiredBatches(requestTimeout, time.milliseconds());
+ 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.abortExpiredBatches(requestTimeout, time.milliseconds());
+ expiredBatches = accum.expiredBatches(requestTimeout, time.milliseconds());
assertEquals("The batch should be expired when the partition is not muted.", 1, expiredBatches.size());
}
@Test
- public void testAppendInExpiryCallback() throws InterruptedException {
- long retryBackoffMs = 100L;
- long lingerMs = 3000L;
- int requestTimeout = 60;
- int messagesPerBatch = expectedNumAppends(1024);
-
- final RecordAccumulator accum = new RecordAccumulator(1024 + DefaultRecordBatch.RECORD_BATCH_OVERHEAD, 10 * 1024,
- CompressionType.NONE, lingerMs, retryBackoffMs, metrics, time, new ApiVersions(), null);
- final AtomicInteger expiryCallbackCount = new AtomicInteger();
- final AtomicReference<Exception> unexpectedException = new AtomicReference<>();
- Callback callback = new Callback() {
- @Override
- public void onCompletion(RecordMetadata metadata, Exception exception) {
- if (exception instanceof TimeoutException) {
- expiryCallbackCount.incrementAndGet();
- try {
- accum.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs);
- } catch (InterruptedException e) {
- throw new RuntimeException("Unexpected interruption", e);
- }
- } else if (exception != null)
- unexpectedException.compareAndSet(null, exception);
- }
- };
-
- for (int i = 0; i < messagesPerBatch + 1; i++)
- accum.append(tp1, 0L, key, value, null, callback, maxBlockTimeMs);
-
- assertEquals(2, accum.batches().get(tp1).size());
- assertTrue("First batch not full", accum.batches().get(tp1).peekFirst().isFull());
-
- // Advance the clock to expire the first batch.
- time.sleep(requestTimeout + 1);
- List<ProducerBatch> expiredBatches = accum.abortExpiredBatches(requestTimeout, time.milliseconds());
- assertEquals("The batch was not expired", 1, expiredBatches.size());
- assertEquals("Callbacks not invoked for expiry", messagesPerBatch, expiryCallbackCount.get());
- assertNull("Unexpected exception", unexpectedException.get());
- assertEquals("Some messages not appended from expiry callbacks", 2, accum.batches().get(tp1).size());
- assertTrue("First batch not full after expiry callbacks with appends", accum.batches().get(tp1).peekFirst().isFull());
- }
-
- @Test
public void testMutedPartitions() throws InterruptedException {
long now = time.milliseconds();
// test case assumes that the records do not fill the batch completely
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
----------------------------------------------------------------------
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 26093d7..5ce5581 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
@@ -22,12 +22,14 @@ import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.NetworkClient;
import org.apache.kafka.clients.NodeApiVersions;
+import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.MetricName;
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.TimeoutException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.MetricConfig;
@@ -40,6 +42,7 @@ import org.apache.kafka.common.record.CompressionRatioEstimator;
import org.apache.kafka.common.record.CompressionType;
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.record.MutableRecordBatch;
+import org.apache.kafka.common.record.Record;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.AddPartitionsToTxnResponse;
@@ -67,6 +70,9 @@ import java.util.List;
import java.util.Map;
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.assertNull;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -348,6 +354,50 @@ public class SenderTest {
}
}
+ @Test
+ public void testAppendInExpiryCallback() throws InterruptedException {
+ int messagesPerBatch = 10;
+ final AtomicInteger expiryCallbackCount = new AtomicInteger(0);
+ final AtomicReference<Exception> unexpectedException = new AtomicReference<>();
+ final byte[] key = "key".getBytes();
+ final byte[] value = "value".getBytes();
+ final long maxBlockTimeMs = 1000;
+ Callback callback = new Callback() {
+ @Override
+ public void onCompletion(RecordMetadata metadata, Exception exception) {
+ if (exception instanceof TimeoutException) {
+ expiryCallbackCount.incrementAndGet();
+ try {
+ accumulator.append(tp1, 0L, key, value, Record.EMPTY_HEADERS, null, maxBlockTimeMs);
+ } catch (InterruptedException e) {
+ throw new RuntimeException("Unexpected interruption", e);
+ }
+ } else if (exception != null)
+ unexpectedException.compareAndSet(null, exception);
+ }
+ };
+
+ for (int i = 0; i < messagesPerBatch; i++)
+ accumulator.append(tp1, 0L, key, value, null, callback, maxBlockTimeMs);
+
+ // Advance the clock to expire the first batch.
+ 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 batch, but we expire it instead without sending anything.
+
+ assertEquals("Callbacks not invoked for expiry", messagesPerBatch, expiryCallbackCount.get());
+ assertNull("Unexpected exception", unexpectedException.get());
+ // Make sure that the reconds were appended back to the batch.
+ assertTrue(accumulator.batches().containsKey(tp1));
+ assertEquals(1, accumulator.batches().get(tp1).size());
+ assertEquals(messagesPerBatch, accumulator.batches().get(tp1).peekFirst().recordCount);
+ }
+
/**
* Tests that topics are added to the metadata list when messages are available to send
* and expired if not used during a metadata refresh interval.
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
----------------------------------------------------------------------
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 55c9241..22afcea 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
@@ -28,6 +28,7 @@ import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.GroupAuthorizationException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.TopicAuthorizationException;
import org.apache.kafka.common.errors.TransactionalIdAuthorizationException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -1361,6 +1362,143 @@ public class TransactionManagerTest {
assertTrue(transactionManager.hasAbortableError());
}
+ @Test
+ public void testRaiseErrorWhenNoPartitionsPendingOnDrain() throws InterruptedException {
+ final long pid = 13131L;
+ final short epoch = 1;
+ doInitTransactions(pid, epoch);
+ transactionManager.beginTransaction();
+ // Don't execute transactionManager.maybeAddPartitionToTransaction(tp0). This should result in an error on drain.
+ accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
+ "value".getBytes(), Record.EMPTY_HEADERS, null, MAX_BLOCK_TIMEOUT);
+ Node node1 = new Node(0, "localhost", 1111);
+ PartitionInfo part1 = new PartitionInfo(topic, 0, node1, null, null);
+
+ Cluster cluster = new Cluster(null, Arrays.asList(node1), Arrays.asList(part1),
+ Collections.<String>emptySet(), Collections.<String>emptySet());
+ Set<Node> nodes = new HashSet<>();
+ nodes.add(node1);
+ Map<Integer, List<ProducerBatch>> drainedBatches = accumulator.drain(cluster, nodes, Integer.MAX_VALUE,
+ time.milliseconds());
+
+ // We shouldn't drain batches which haven't been added to the transaction yet.
+ assertTrue(drainedBatches.containsKey(node1.id()));
+ assertTrue(drainedBatches.get(node1.id()).isEmpty());
+ }
+
+ @Test
+ public void testTransitionToAbortableErrorOnBatchExpiry() throws InterruptedException, ExecutionException {
+ final long pid = 13131L;
+ final short epoch = 1;
+
+ doInitTransactions(pid, epoch);
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartitionToTransaction(tp0);
+
+ Future<RecordMetadata> 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));
+ assertFalse(responseFuture.isDone());
+
+ // 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);
+ }
+ assertTrue(transactionManager.hasAbortableError());
+ }
+
+ @Test
+ public void testDropCommitOnBatchExpiry() throws InterruptedException, ExecutionException {
+ final long pid = 13131L;
+ final short epoch = 1;
+
+ doInitTransactions(pid, epoch);
+
+ transactionManager.beginTransaction();
+ transactionManager.maybeAddPartitionToTransaction(tp0);
+
+ Future<RecordMetadata> 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));
+ assertFalse(responseFuture.isDone());
+
+ TransactionalRequestResult commitResult = transactionManager.beginCommittingTransaction();
+
+ // 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()); // the commit shouldn't be completed without being sent since the produce request failed.
+
+ assertTrue(commitResult.isCompleted());
+ assertFalse(commitResult.isSuccessful()); // the commit shouldn't succeed since the produce request failed.
+
+ assertTrue(transactionManager.hasAbortableError());
+ assertTrue(transactionManager.hasOngoingTransaction());
+ assertFalse(transactionManager.isCompletingTransaction());
+ assertTrue(transactionManager.transactionContainsPartition(tp0));
+
+ TransactionalRequestResult abortResult = transactionManager.beginAbortingTransaction();
+
+ prepareEndTxnResponse(Errors.NONE, TransactionResult.ABORT, pid, epoch);
+
+ sender.run(time.milliseconds()); // send the abort.
+
+ assertTrue(abortResult.isCompleted());
+ assertTrue(abortResult.isSuccessful());
+ assertFalse(transactionManager.hasOngoingTransaction());
+ assertFalse(transactionManager.transactionContainsPartition(tp0));
+ }
+
private void verifyAddPartitionsFailsWithPartitionLevelError(final Errors error) throws InterruptedException {
final long pid = 1L;
final short epoch = 1;
http://git-wip-us.apache.org/repos/asf/kafka/blob/79db393f/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java
----------------------------------------------------------------------
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 8316785..d748ce4 100644
--- a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java
+++ b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java
@@ -29,10 +29,12 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
@@ -180,6 +182,16 @@ public class TransactionalMessageCopier {
return positions;
}
+ private static void resetToLastCommittedPositions(KafkaConsumer<String, String> consumer) {
+ for (TopicPartition topicPartition : consumer.assignment()) {
+ OffsetAndMetadata offsetAndMetadata = consumer.committed(topicPartition);
+ if (offsetAndMetadata != null)
+ consumer.seek(topicPartition, offsetAndMetadata.offset());
+ else
+ consumer.seekToBeginning(Collections.singletonList(topicPartition));
+ }
+ }
+
private static long messagesRemaining(KafkaConsumer<String, String> consumer, TopicPartition partition) {
long currentPosition = consumer.position(partition);
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(Arrays.asList(partition));
@@ -235,11 +247,9 @@ public class TransactionalMessageCopier {
producer.initTransactions();
-
final AtomicBoolean isShuttingDown = new AtomicBoolean(false);
final AtomicLong remainingMessages = new AtomicLong(maxMessages);
final AtomicLong numMessagesProcessed = new AtomicLong(0);
- int exitCode = 0;
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
@@ -260,18 +270,22 @@ public class TransactionalMessageCopier {
break;
int messagesInCurrentTransaction = 0;
long numMessagesForNextTransaction = Math.min(numMessagesPerTransaction, remainingMessages.get());
- producer.beginTransaction();
-
- while (messagesInCurrentTransaction < numMessagesForNextTransaction) {
- ConsumerRecords<String, String> records = consumer.poll(200L);
- for (ConsumerRecord<String, String> record : records) {
- producer.send(producerRecordFromConsumerRecord(outputTopic, record));
- messagesInCurrentTransaction++;
+ try {
+ producer.beginTransaction();
+ while (messagesInCurrentTransaction < numMessagesForNextTransaction) {
+ ConsumerRecords<String, String> records = consumer.poll(200L);
+ for (ConsumerRecord<String, String> record : records) {
+ producer.send(producerRecordFromConsumerRecord(outputTopic, record));
+ messagesInCurrentTransaction++;
+ }
}
+ producer.sendOffsetsToTransaction(consumerPositions(consumer), consumerGroup);
+ producer.commitTransaction();
+ remainingMessages.set(maxMessages - numMessagesProcessed.addAndGet(messagesInCurrentTransaction));
+ } catch (KafkaException e) {
+ producer.abortTransaction();
+ resetToLastCommittedPositions(consumer);
}
- producer.sendOffsetsToTransaction(consumerPositions(consumer), consumerGroup);
- producer.commitTransaction();
- remainingMessages.set(maxMessages - numMessagesProcessed.addAndGet(messagesInCurrentTransaction));
}
} finally {
producer.close();
@@ -279,6 +293,6 @@ public class TransactionalMessageCopier {
consumer.close();
}
}
- System.exit(exitCode);
+ System.exit(0);
}
}