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 2 commits
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
93 changes: 49 additions & 44 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 @@ -46,6 +46,7 @@
import org.apache.kafka.common.record.MemoryRecords;
import org.apache.kafka.common.record.RecordBatch;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.FindCoordinatorRequest;
import org.apache.kafka.common.requests.InitProducerIdRequest;
import org.apache.kafka.common.requests.InitProducerIdResponse;
import org.apache.kafka.common.requests.ProduceRequest;
Expand Down Expand Up @@ -302,9 +303,7 @@ 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()) {
// as long as there are outstanding transactional requests, we simply wait for them to return
client.poll(retryBackoffMs, time.milliseconds());
} else if (maybeSendAndPollTransactionalRequest()) {
return;
}

Expand Down Expand Up @@ -412,7 +411,16 @@ private long sendProducerData(long now) {
return pollTimeout;
}

private boolean maybeSendTransactionalRequest() {
/**
* Returns true if a transactional request is sent or polled, or if a FindCoordinator request is enqueued
*/
private boolean maybeSendAndPollTransactionalRequest() {
if (transactionManager.hasInFlightTransactionalRequest()) {
// as long as there are outstanding transactional requests, we simply wait for them to return
client.poll(retryBackoffMs, time.milliseconds());
return true;
}

if (transactionManager.isCompleting() && accumulator.hasIncomplete()) {
if (transactionManager.isAborting())
accumulator.abortUndrainedBatches(new KafkaException("Failing batch since transaction was aborted"));
Expand All @@ -429,48 +437,42 @@ private boolean maybeSendTransactionalRequest() {
return false;

AbstractRequest.Builder<?> requestBuilder = nextRequestHandler.requestBuilder();
while (!forceClose) {
Node targetNode = null;
try {
if (nextRequestHandler.needsCoordinator()) {
targetNode = transactionManager.coordinator(nextRequestHandler.coordinatorType());
if (targetNode == null) {
transactionManager.lookupCoordinator(nextRequestHandler);
break;
}
if (!NetworkClientUtils.awaitReady(client, targetNode, time, requestTimeoutMs)) {
transactionManager.lookupCoordinator(nextRequestHandler);
break;
}
} else {
targetNode = awaitLeastLoadedNodeReady(requestTimeoutMs);
}

if (targetNode != null) {
if (nextRequestHandler.isRetry())
time.sleep(nextRequestHandler.retryBackoffMs());
long currentTimeMs = time.milliseconds();
ClientRequest clientRequest = client.newClientRequest(
targetNode.idString(), requestBuilder, currentTimeMs, true, requestTimeoutMs, nextRequestHandler);
log.debug("Sending transactional request {} to node {}", requestBuilder, targetNode);
client.send(clientRequest, currentTimeMs);
transactionManager.setInFlightCorrelationId(clientRequest.correlationId());
return true;
}
} catch (IOException e) {
log.debug("Disconnect from {} while trying to send request {}. Going " +
"to back off and retry.", targetNode, requestBuilder, e);
if (nextRequestHandler.needsCoordinator()) {
// We break here so that we pick up the FindCoordinator request immediately.
transactionManager.lookupCoordinator(nextRequestHandler);
break;
}
Node targetNode = null;
try {
targetNode = awaitLeastLoadedNodeReady(requestTimeoutMs, nextRequestHandler.coordinatorType());
if (targetNode == null) {
lookupCoordinatorAndRetry(nextRequestHandler);
return true;
}

if (nextRequestHandler.isRetry())
time.sleep(nextRequestHandler.retryBackoffMs());
long currentTimeMs = time.milliseconds();
ClientRequest clientRequest = client.newClientRequest(
targetNode.idString(), requestBuilder, currentTimeMs, true, requestTimeoutMs, nextRequestHandler);
log.debug("Sending transactional request {} to node {}", requestBuilder, targetNode);
client.send(clientRequest, currentTimeMs);
transactionManager.setInFlightCorrelationId(clientRequest.correlationId());
client.poll(retryBackoffMs, time.milliseconds());
return true;
} catch (IOException e) {
log.debug("Disconnect from {} while trying to send request {}. Going " +
"to back off and retry.", targetNode, requestBuilder, e);
// We break here so that we pick up the FindCoordinator request immediately.
lookupCoordinatorAndRetry(nextRequestHandler);
return true;
}
}

private void lookupCoordinatorAndRetry(TransactionManager.TxnRequestHandler nextRequestHandler) {
if (!nextRequestHandler.needsCoordinator()) {
// For non-coordinator requests, sleep here to prevent a tight loop when no node is available
time.sleep(retryBackoffMs);
metadata.requestUpdate();
}

transactionManager.lookupCoordinator(nextRequestHandler);

@hachikuji hachikuji Jul 18, 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.

It reads a bit strange to fall through to lookupCoordinator if we know the request doesn't need the coordinator. Maybe clearer with a slight restructure:

transactionManager.retry(nextRequestHandler);

if (nextRequestHandler.needsCoordinator()) {
  transactionManager.lookupCoordinator(nextRequestHandler);
} else {
  // For non-coordinator requests, sleep here to prevent a tight loop when no node is available
  time.sleep(retryBackoffMs);
  metadata.requestUpdate();	            
}

transactionManager.retry(nextRequestHandler);
return true;
}

private void maybeAbortBatches(RuntimeException exception) {
Expand Down Expand Up @@ -513,8 +515,11 @@ private ClientResponse sendAndAwaitInitProducerIdRequest(Node node) throws IOExc
return NetworkClientUtils.sendAndReceive(client, request, time);
}

private Node awaitLeastLoadedNodeReady(long remainingTimeMs) throws IOException {
Node node = client.leastLoadedNode(time.milliseconds());
private Node awaitLeastLoadedNodeReady(long remainingTimeMs, FindCoordinatorRequest.CoordinatorType coordinatorType) throws IOException {

@hachikuji hachikuji Jul 18, 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.

nit: the name should be changed since we do not always use the least loaded node.

I noticed we are always using request timeout for the argument. Maybe we can drop the argument?

Node node = coordinatorType != null ?
transactionManager.coordinator(coordinatorType) :
client.leastLoadedNode(time.milliseconds());

if (node != null && NetworkClientUtils.awaitReady(client, node, time, remainingTimeMs)) {
return node;
}
Expand All @@ -525,7 +530,7 @@ private void maybeWaitForProducerId() {
while (!forceClose && !transactionManager.hasProducerId() && !transactionManager.hasError()) {
Node node = null;
try {
node = awaitLeastLoadedNodeReady(requestTimeoutMs);
node = awaitLeastLoadedNodeReady(requestTimeoutMs, null);
if (node != null) {
ClientResponse response = sendAndAwaitInitProducerIdRequest(node);
InitProducerIdResponse initProducerIdResponse = (InitProducerIdResponse) response.responseBody();
Expand Down
6 changes: 4 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -944,6 +944,9 @@ private void enqueueRequest(TxnRequestHandler requestHandler) {
}

private void lookupCoordinator(FindCoordinatorRequest.CoordinatorType type, String coordinatorKey) {
if (type == null)
return;

switch (type) {
case GROUP:
consumerGroupCoordinator = null;
Expand Down Expand Up @@ -1057,8 +1060,7 @@ public void onComplete(ClientResponse response) {
clearInFlightCorrelationId();
if (response.wasDisconnected()) {
log.debug("Disconnected from {}. Will retry.", response.destination());
if (this.needsCoordinator())
lookupCoordinator(this.coordinatorType(), this.coordinatorKey());
lookupCoordinator(this.coordinatorType(), this.coordinatorKey());
reenqueue();
} else if (response.versionMismatch() != null) {
fatalError(response.versionMismatch());
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