Skip to content

KAFKA-8635: Skip client poll in Sender loop when no request is sent #7085

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
hachikuji merged 3 commits into from
Jul 18, 2019
Merged

KAFKA-8635: Skip client poll in Sender loop when no request is sent #7085

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ddd2d3f
KAFKA-8635: Skip client poll in Sender loop when no request is sent
bob-barrett Jul 12, 2019
2da11bc
PR feedback: consolidate awaitReady path, move in-flight check into h…
bob-barrett Jul 16, 2019
0cd3376
Rename awaitLeastLoadedNode, check before calling lookupCoordinator
bob-barrett Jul 18, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,12 +302,22 @@ void runOnce() {
transactionManager.transitionToFatalError(
new KafkaException("The client hasn't received acknowledgment for " +
"some previously sent messages and can no longer retry them. It isn't safe to continue."));
} else if (transactionManager.hasInFlightTransactionalRequest() || maybeSendTransactionalRequest()) {
} else if (transactionManager.hasInFlightTransactionalRequest()) {
// as long as there are outstanding transactional requests, we simply wait for them to return
client.poll(retryBackoffMs, time.milliseconds());
return;
}

maybeBeginFlush();
TransactionManager.TxnRequestHandler nextRequestHandler = transactionManager.nextRequestHandler(accumulator.hasIncomplete());
if (nextRequestHandler != null) {
if (maybeSendTransactionalRequest(nextRequestHandler)) {
client.poll(retryBackoffMs, time.milliseconds());

@hachikuji hachikuji Jul 13, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This logic is a bit awkward. I guess it's because client.poll is disconnected from the send logic. I wonder if it would be reasonable to create a method like maybePollTransactionalRequest which handles both the inflight check, sending the next request, and calling client.poll if necessary.

}

return;
}

// do not continue sending if the transaction manager is in a failed state or if there
// is no producer id (for the idempotent case).
if (transactionManager.hasFatalError() || !transactionManager.hasProducerId()) {
Expand Down Expand Up @@ -412,7 +422,7 @@ private long sendProducerData(long now) {
return pollTimeout;
}

private boolean maybeSendTransactionalRequest() {
private void maybeBeginFlush() {
if (transactionManager.isCompleting() && accumulator.hasIncomplete()) {
if (transactionManager.isAborting())
accumulator.abortUndrainedBatches(new KafkaException("Failing batch since transaction was aborted"));
Expand All @@ -423,11 +433,12 @@ private boolean maybeSendTransactionalRequest() {
if (!accumulator.flushInProgress())
accumulator.beginFlush();
}
}

TransactionManager.TxnRequestHandler nextRequestHandler = transactionManager.nextRequestHandler(accumulator.hasIncomplete());
if (nextRequestHandler == null)
return false;

/**
* Enqueue a FindCoordinatorRequest if needed, otherwise send the next request
*/
private boolean maybeSendTransactionalRequest(TransactionManager.TxnRequestHandler nextRequestHandler) {
AbstractRequest.Builder<?> requestBuilder = nextRequestHandler.requestBuilder();
while (!forceClose) {

@hachikuji hachikuji Jul 13, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not necessarily something we have to do here, but I think we should be able to get rid of this loop and just rely on the next iteration of runOnce() to handle retries.

Node targetNode = null;

@hachikuji hachikuji Jul 13, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor simplification we can do below:

if (targetNode == null || !awaitReady(client, targetNode, time, requestTimeoutMs)) {
  transactionManager.lookupCoordinator(nextRequestHandler);
  break;                        
}

@hachikuji hachikuji Jul 13, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A helper for awaitReady might be useful as well. Might be a chance to consolidate the awaitLeastLoadedNodeReady path.

Expand Down Expand Up @@ -470,7 +481,7 @@ private boolean maybeSendTransactionalRequest() {
metadata.requestUpdate();
}
transactionManager.retry(nextRequestHandler);
return true;
return false;
}

private void maybeAbortBatches(RuntimeException exception) {
Expand Down
44 changes: 30 additions & 14 deletions clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,8 @@
import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

public class SenderTest {
private static final int MAX_REQUEST_SIZE = 1024 * 1024;
Expand All @@ -121,6 +123,7 @@ public class SenderTest {
private static final double EPS = 0.0001;
private static final int MAX_BLOCK_TIMEOUT = 1000;
private static final int REQUEST_TIMEOUT = 1000;
private static final long RETRY_BACKOFF_MS = 50;

private TopicPartition tp0 = new TopicPartition("test", 0);
private TopicPartition tp1 = new TopicPartition("test", 1);
Expand Down Expand Up @@ -315,7 +318,7 @@ public void testSenderMetricsTemplates() throws Exception {
metrics = new Metrics(new MetricConfig().tags(clientTags));
SenderMetricsRegistry metricsRegistry = new SenderMetricsRegistry(metrics);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
1, metricsRegistry, time, REQUEST_TIMEOUT, 50, null, apiVersions);
1, metricsRegistry, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null, apiVersions);

// Append a message so that topic metrics are created
accumulator.append(tp0, 0L, "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT);
Expand Down Expand Up @@ -343,7 +346,7 @@ public void testRetries() throws Exception {
SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
try {
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, null, apiVersions);
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null, apiVersions);
// do a successful retry
Future<RecordMetadata> future = accumulator.append(tp0, 0L, "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
sender.runOnce(); // connect
Expand Down Expand Up @@ -401,7 +404,7 @@ public void testSendInOrder() throws Exception {

try {
Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries,
senderMetrics, time, REQUEST_TIMEOUT, 50, null, apiVersions);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null, apiVersions);
// Create a two broker cluster, with partition 0 on broker 0 and partition 1 on broker 1
MetadataResponse metadataUpdate1 = TestUtils.metadataUpdateWith(2, Collections.singletonMap("test", 2));
client.prepareMetadataUpdate(metadataUpdate1);
Expand Down Expand Up @@ -1172,7 +1175,7 @@ public void testResetOfProducerStateShouldAllowQueuedBatchesToDrain() throws Exc
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);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> failedResponse = accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
"value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
Expand Down Expand Up @@ -1214,7 +1217,7 @@ public void testCloseWithProducerIdReset() throws Exception {
SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, 10,
senderMetrics, time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> failedResponse = accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
"value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
Expand Down Expand Up @@ -1253,7 +1256,7 @@ public void testForceCloseWithProducerIdReset() throws Exception {
SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, 10,
senderMetrics, time, REQUEST_TIMEOUT, 50, transactionManager, apiVersions);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> failedResponse = accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
"value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
Expand Down Expand Up @@ -1289,7 +1292,7 @@ public void testBatchesDrainedWithOldProducerIdShouldFailWithOutOfOrderSequenceO
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);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> failedResponse = accumulator.append(tp0, time.milliseconds(), "key".getBytes(),
"value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
Expand Down Expand Up @@ -1778,7 +1781,7 @@ public void testSequenceNumberIncrement() throws InterruptedException {
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);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> responseFuture = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
client.prepareResponse(new MockClient.RequestMatcher() {
Expand Down Expand Up @@ -1820,7 +1823,7 @@ public void testAbortRetryWhenProducerIdChanges() throws InterruptedException {
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);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> responseFuture = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
sender.runOnce(); // connect.
Expand Down Expand Up @@ -1859,7 +1862,7 @@ public void testResetWhenOutOfOrderSequenceReceived() throws InterruptedExceptio
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);
senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

Future<RecordMetadata> responseFuture = accumulator.append(tp0, time.milliseconds(), "key".getBytes(), "value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
sender.runOnce(); // connect.
Expand Down Expand Up @@ -2205,7 +2208,7 @@ public void testTransactionalRequestsSentOnShutdown() {
try {
TransactionManager txnManager = new TransactionManager(logContext, "testTransactionalRequestsSentOnShutdown", 6000, 100);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, txnManager, apiVersions);
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, txnManager, apiVersions);

ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
TopicPartition tp = new TopicPartition("testTransactionalRequestsSentOnShutdown", 1);
Expand Down Expand Up @@ -2237,7 +2240,7 @@ public void testIncompleteTransactionAbortOnShutdown() {
try {
TransactionManager txnManager = new TransactionManager(logContext, "testIncompleteTransactionAbortOnShutdown", 6000, 100);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, txnManager, apiVersions);
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, txnManager, apiVersions);

ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
TopicPartition tp = new TopicPartition("testIncompleteTransactionAbortOnShutdown", 1);
Expand Down Expand Up @@ -2268,7 +2271,7 @@ public void testForceShutdownWithIncompleteTransaction() {
try {
TransactionManager txnManager = new TransactionManager(logContext, "testForceShutdownWithIncompleteTransaction", 6000, 100);
Sender sender = new Sender(logContext, client, metadata, this.accumulator, false, MAX_REQUEST_SIZE, ACKS_ALL,
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, 50, txnManager, apiVersions);
maxRetries, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, txnManager, apiVersions);

ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
TopicPartition tp = new TopicPartition("testForceShutdownWithIncompleteTransaction", 1);
Expand All @@ -2293,6 +2296,19 @@ public void testForceShutdownWithIncompleteTransaction() {
}
}

@Test
public void testDoNotPollWhenNoRequestSent() {
client = spy(new MockClient(time, metadata));

TransactionManager txnManager = new TransactionManager(logContext, "testDoNotPollWhenNoRequestSent", 6000, 100);
ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(123456L, (short) 0);
setupWithTransactionState(txnManager);
doInitTransactions(txnManager, producerIdAndEpoch);

// doInitTransactions calls sender.doOnce three times, only two requests are sent, so we should only poll twice
verify(client, times(2)).poll(eq(RETRY_BACKOFF_MS), anyLong());
}

class AssertEndTxnRequestMatcher implements MockClient.RequestMatcher {

private TransactionResult requiredResult;
Expand Down Expand Up @@ -2418,7 +2434,7 @@ private void setupWithTransactionState(TransactionManager transactionManager, bo
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);
Integer.MAX_VALUE, this.senderMetricsRegistry, this.time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

metadata.add("test");
this.client.updateMetadata(TestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
Expand Down