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 2020/10/21 19:20:31 UTC
[kafka] branch 2.7 updated: KAFKA-10520;
Ensure transactional producers poll if leastLoadedNode not
available with max.in.flight=1 (#9406)
This is an automated email from the ASF dual-hosted git repository.
jgus pushed a commit to branch 2.7
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/2.7 by this push:
new ab65b8b KAFKA-10520; Ensure transactional producers poll if leastLoadedNode not available with max.in.flight=1 (#9406)
ab65b8b is described below
commit ab65b8b8a307384d1eddeb457852039781d328e6
Author: Rajini Sivaram <ra...@googlemail.com>
AuthorDate: Wed Oct 21 20:07:27 2020 +0100
KAFKA-10520; Ensure transactional producers poll if leastLoadedNode not available with max.in.flight=1 (#9406)
We currently stop polling in `Sender` in a transactional producer if there is only one broker in the bootstrap server list and `max.in.flight.requests.per.connection=1` and Metadata response is pending when InitProducerId request is ready to be sent. In this scenario, we attempt to send FindCoordinator to `leastLoadedNode`, but since that is blocked due to `max.in.flight=1` as a result of the pending metadata response, we never unblock unless we poll. This PR ensures we poll in this case.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Jason Gustafson <ja...@confluent.io>, David Jacot <dj...@confluent.io>
---
.../kafka/clients/producer/internals/Sender.java | 31 +++--
.../clients/producer/internals/SenderTest.java | 136 ++++++++++++++++++++-
.../api/TransactionsWithMaxInFlightOneTest.scala | 131 ++++++++++++++++++++
.../test/scala/unit/kafka/utils/TestUtils.scala | 4 +-
4 files changed, 289 insertions(+), 13 deletions(-)
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 958f17f..19b2cb1 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
@@ -444,10 +444,25 @@ public class Sender implements Runnable {
AbstractRequest.Builder<?> requestBuilder = nextRequestHandler.requestBuilder();
Node targetNode = null;
try {
- targetNode = awaitNodeReady(nextRequestHandler.coordinatorType());
- if (targetNode == null) {
+ FindCoordinatorRequest.CoordinatorType coordinatorType = nextRequestHandler.coordinatorType();
+ targetNode = coordinatorType != null ?
+ transactionManager.coordinator(coordinatorType) :
+ client.leastLoadedNode(time.milliseconds());
+ if (targetNode != null) {
+ if (!awaitNodeReady(targetNode, coordinatorType)) {
+ log.trace("Target node {} not ready within request timeout, will retry when node is ready.", targetNode);
+ maybeFindCoordinatorAndRetry(nextRequestHandler);
+ return true;
+ }
+ } else if (coordinatorType != null) {
+ log.trace("Coordinator not known for {}, will retry {} after finding coordinator.", coordinatorType, requestBuilder.apiKey());
maybeFindCoordinatorAndRetry(nextRequestHandler);
return true;
+ } else {
+ log.trace("No nodes available to send requests, will poll and retry when until a node is ready.");
+ transactionManager.retry(nextRequestHandler);
+ client.poll(retryBackoffMs, time.milliseconds());
+ return true;
}
if (nextRequestHandler.isRetry())
@@ -512,21 +527,17 @@ public class Sender implements Runnable {
return running;
}
- private Node awaitNodeReady(FindCoordinatorRequest.CoordinatorType coordinatorType) throws IOException {
- Node node = coordinatorType != null ?
- transactionManager.coordinator(coordinatorType) :
- client.leastLoadedNode(time.milliseconds());
-
- if (node != null && NetworkClientUtils.awaitReady(client, node, time, requestTimeoutMs)) {
+ private boolean awaitNodeReady(Node node, FindCoordinatorRequest.CoordinatorType coordinatorType) throws IOException {
+ if (NetworkClientUtils.awaitReady(client, node, time, requestTimeoutMs)) {
if (coordinatorType == FindCoordinatorRequest.CoordinatorType.TRANSACTION) {
// Indicate to the transaction manager that the coordinator is ready, allowing it to check ApiVersions
// This allows us to bump transactional epochs even if the coordinator is temporarily unavailable at
// the time when the abortable error is handled
transactionManager.handleCoordinatorReady();
}
- return node;
+ return true;
}
- return null;
+ return false;
}
/**
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 85d8db8..771f699 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
@@ -19,12 +19,15 @@ package org.apache.kafka.clients.producer.internals;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientDnsLookup;
import org.apache.kafka.clients.ClientRequest;
+import org.apache.kafka.clients.ClientResponse;
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.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.TransactionAbortedException;
+import org.apache.kafka.common.requests.FindCoordinatorRequest.CoordinatorType;
+import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.utils.ProducerIdAndEpoch;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.Cluster;
@@ -103,6 +106,7 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;
import static org.mockito.AdditionalMatchers.geq;
@@ -532,6 +536,90 @@ public class SenderTest {
assertEquals((short) 0, transactionManager.producerIdAndEpoch().epoch);
}
+ /**
+ * Verifies that InitProducerId of transactional producer succeeds even if metadata requests
+ * are pending with only one bootstrap node available and maxInFlight=1, where multiple
+ * polls are necessary to send requests.
+ */
+ @Test
+ public void testInitProducerIdWithMaxInFlightOne() throws Exception {
+ final long producerId = 123456L;
+ createMockClientWithMaxFlightOneMetadataPending();
+
+ // Initialize transaction manager. InitProducerId will be queued up until metadata response
+ // is processed and FindCoordinator can be sent to `leastLoadedNode`.
+ TransactionManager transactionManager = new TransactionManager(new LogContext(), "testInitProducerIdWithPendingMetadataRequest",
+ 60000, 100L, new ApiVersions(), false);
+ setupWithTransactionState(transactionManager, false, null, false);
+ ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);
+ transactionManager.initializeTransactions();
+ sender.runOnce();
+
+ // Process metadata response, prepare FindCoordinator and InitProducerId responses.
+ // Verify producerId after the sender is run to process responses.
+ MetadataResponse metadataUpdate = TestUtils.metadataUpdateWith(1, Collections.emptyMap());
+ client.respond(metadataUpdate);
+ prepareFindCoordinatorResponse(Errors.NONE);
+ prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
+ waitForProducerId(transactionManager, producerIdAndEpoch);
+ }
+
+ /**
+ * Verifies that InitProducerId of idempotent producer succeeds even if metadata requests
+ * are pending with only one bootstrap node available and maxInFlight=1, where multiple
+ * polls are necessary to send requests.
+ */
+ @Test
+ public void testIdempotentInitProducerIdWithMaxInFlightOne() throws Exception {
+ final long producerId = 123456L;
+ createMockClientWithMaxFlightOneMetadataPending();
+
+ // Initialize transaction manager. InitProducerId will be queued up until metadata response
+ // is processed.
+ TransactionManager transactionManager = createTransactionManager();
+ setupWithTransactionState(transactionManager, false, null, false);
+ ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);
+
+ // Process metadata and InitProducerId responses.
+ // Verify producerId after the sender is run to process responses.
+ MetadataResponse metadataUpdate = TestUtils.metadataUpdateWith(1, Collections.emptyMap());
+ client.respond(metadataUpdate);
+ sender.runOnce();
+ sender.runOnce();
+ client.respond(initProducerIdResponse(producerIdAndEpoch.producerId, producerIdAndEpoch.epoch, Errors.NONE));
+ waitForProducerId(transactionManager, producerIdAndEpoch);
+ }
+
+ /**
+ * Tests the code path where the target node to send FindCoordinator or InitProducerId
+ * is not ready.
+ */
+ @Test
+ public void testNodeNotReady() throws Exception {
+ final long producerId = 123456L;
+ time = new MockTime(10);
+ client = new MockClient(time, metadata);
+
+ TransactionManager transactionManager = new TransactionManager(new LogContext(), "testNodeNotReady",
+ 60000, 100L, new ApiVersions(), false);
+ setupWithTransactionState(transactionManager, false, null, true);
+ ProducerIdAndEpoch producerIdAndEpoch = new ProducerIdAndEpoch(producerId, (short) 0);
+ transactionManager.initializeTransactions();
+ sender.runOnce();
+
+ Node node = metadata.fetch().nodes().get(0);
+ client.delayReady(node, REQUEST_TIMEOUT + 20);
+ prepareFindCoordinatorResponse(Errors.NONE);
+ sender.runOnce();
+ sender.runOnce();
+ assertNotNull("Coordinator not found", transactionManager.coordinator(CoordinatorType.TRANSACTION));
+
+ client.throttle(node, REQUEST_TIMEOUT + 20);
+ prepareFindCoordinatorResponse(Errors.NONE);
+ prepareInitProducerResponse(Errors.NONE, producerIdAndEpoch.producerId, producerIdAndEpoch.epoch);
+ waitForProducerId(transactionManager, producerIdAndEpoch);
+ }
+
@Test
public void testClusterAuthorizationExceptionInInitProducerIdRequest() throws Exception {
final long producerId = 343434L;
@@ -2578,10 +2666,14 @@ public class SenderTest {
}
private void setupWithTransactionState(TransactionManager transactionManager) {
- setupWithTransactionState(transactionManager, false, null);
+ setupWithTransactionState(transactionManager, false, null, true);
}
private void setupWithTransactionState(TransactionManager transactionManager, boolean guaranteeOrder, BufferPool customPool) {
+ setupWithTransactionState(transactionManager, guaranteeOrder, customPool, true);
+ }
+
+ private void setupWithTransactionState(TransactionManager transactionManager, boolean guaranteeOrder, BufferPool customPool, boolean updateMetadata) {
int deliveryTimeoutMs = 1500;
long totalSize = 1024 * 1024;
String metricGrpName = "producer-metrics";
@@ -2596,7 +2688,8 @@ public class SenderTest {
Integer.MAX_VALUE, this.senderMetricsRegistry, this.time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);
metadata.add("test", time.milliseconds());
- this.client.updateMetadata(TestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
+ if (updateMetadata)
+ this.client.updateMetadata(TestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 2)));
}
private void assertSendFailure(Class<? extends RuntimeException> expectedError) throws Exception {
@@ -2667,4 +2760,43 @@ public class SenderTest {
}
}
+ private void createMockClientWithMaxFlightOneMetadataPending() {
+ client = new MockClient(time, metadata) {
+ volatile boolean canSendMore = true;
+ @Override
+ public Node leastLoadedNode(long now) {
+ for (Node node : metadata.fetch().nodes()) {
+ if (isReady(node, now) && canSendMore)
+ return node;
+ }
+ return null;
+ }
+
+ @Override
+ public List<ClientResponse> poll(long timeoutMs, long now) {
+ canSendMore = inFlightRequestCount() < 1;
+ return super.poll(timeoutMs, now);
+ }
+ };
+
+ // Send metadata request and wait until request is sent. `leastLoadedNode` will be null once
+ // request is in progress since no more requests can be sent to the node. Node will be ready
+ // on the next poll() after response is processed later on in tests which use this method.
+ MetadataRequest.Builder builder = new MetadataRequest.Builder(Collections.emptyList(), false);
+ Node node = metadata.fetch().nodes().get(0);
+ ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true);
+ while (!client.ready(node, time.milliseconds()))
+ client.poll(0, time.milliseconds());
+ client.send(request, time.milliseconds());
+ while (client.leastLoadedNode(time.milliseconds()) != null)
+ client.poll(0, time.milliseconds());
+ }
+
+ private void waitForProducerId(TransactionManager transactionManager, ProducerIdAndEpoch producerIdAndEpoch) {
+ for (int i = 0; i < 5 && !transactionManager.hasProducerId(); i++)
+ sender.runOnce();
+
+ assertTrue(transactionManager.hasProducerId());
+ assertEquals(producerIdAndEpoch, transactionManager.producerIdAndEpoch());
+ }
}
diff --git a/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala b/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala
new file mode 100644
index 0000000..4d00a2c
--- /dev/null
+++ b/core/src/test/scala/integration/kafka/api/TransactionsWithMaxInFlightOneTest.scala
@@ -0,0 +1,131 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package integration.kafka.api
+
+import java.util.Properties
+
+import kafka.integration.KafkaServerTestHarness
+import kafka.server.KafkaConfig
+import kafka.utils.TestUtils
+import kafka.utils.TestUtils.consumeRecords
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.junit.{After, Before, Test}
+import org.junit.Assert.assertEquals
+
+import scala.collection.Seq
+import scala.collection.mutable.Buffer
+import scala.jdk.CollectionConverters._
+
+/**
+ * This is used to test transactions with one broker and `max.in.flight.requests.per.connection=1`.
+ * A single broker is used to verify edge cases where different requests are queued on the same connection.
+ */
+class TransactionsWithMaxInFlightOneTest extends KafkaServerTestHarness {
+ val numServers = 1
+
+ val topic1 = "topic1"
+ val topic2 = "topic2"
+ val numPartitions = 4
+
+ val transactionalProducers = Buffer[KafkaProducer[Array[Byte], Array[Byte]]]()
+ val transactionalConsumers = Buffer[KafkaConsumer[Array[Byte], Array[Byte]]]()
+
+ override def generateConfigs: Seq[KafkaConfig] = {
+ TestUtils.createBrokerConfigs(numServers, zkConnect).map(KafkaConfig.fromProps(_, serverProps()))
+ }
+
+ @Before
+ override def setUp(): Unit = {
+ super.setUp()
+ val topicConfig = new Properties()
+ topicConfig.put(KafkaConfig.MinInSyncReplicasProp, 1.toString)
+ createTopic(topic1, numPartitions, numServers, topicConfig)
+ createTopic(topic2, numPartitions, numServers, topicConfig)
+
+ createTransactionalProducer("transactional-producer")
+ createReadCommittedConsumer("transactional-group")
+ }
+
+ @After
+ override def tearDown(): Unit = {
+ transactionalProducers.foreach(_.close())
+ transactionalConsumers.foreach(_.close())
+ super.tearDown()
+ }
+
+ @Test
+ def testTransactionalProducerSingleBrokerMaxInFlightOne() = {
+ // We want to test with one broker to verify multiple requests queued on a connection
+ assertEquals(1, servers.size)
+
+ val producer = transactionalProducers.head
+ val consumer = transactionalConsumers.head
+
+ producer.initTransactions()
+
+ producer.beginTransaction()
+ producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic2, null, "2", "2", willBeCommitted = false))
+ producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, null, "4", "4", willBeCommitted = false))
+ producer.flush()
+ producer.abortTransaction()
+
+ producer.beginTransaction()
+ producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic1, null, "1", "1", willBeCommitted = true))
+ producer.send(TestUtils.producerRecordWithExpectedTransactionStatus(topic2, null, "3", "3", willBeCommitted = true))
+ producer.commitTransaction()
+
+ consumer.subscribe(List(topic1, topic2).asJava)
+
+ val records = consumeRecords(consumer, 2)
+ records.foreach { record =>
+ TestUtils.assertCommittedAndGetValue(record)
+ }
+ }
+
+ private def serverProps() = {
+ val serverProps = new Properties()
+ serverProps.put(KafkaConfig.AutoCreateTopicsEnableProp, false.toString)
+ serverProps.put(KafkaConfig.OffsetsTopicPartitionsProp, 1.toString)
+ serverProps.put(KafkaConfig.OffsetsTopicReplicationFactorProp, 1.toString)
+ serverProps.put(KafkaConfig.TransactionsTopicPartitionsProp, 1.toString)
+ serverProps.put(KafkaConfig.TransactionsTopicReplicationFactorProp, 1.toString)
+ serverProps.put(KafkaConfig.TransactionsTopicMinISRProp, 1.toString)
+ serverProps.put(KafkaConfig.ControlledShutdownEnableProp, true.toString)
+ serverProps.put(KafkaConfig.UncleanLeaderElectionEnableProp, false.toString)
+ serverProps.put(KafkaConfig.AutoLeaderRebalanceEnableProp, false.toString)
+ serverProps.put(KafkaConfig.GroupInitialRebalanceDelayMsProp, "0")
+ serverProps.put(KafkaConfig.TransactionsAbortTimedOutTransactionCleanupIntervalMsProp, "200")
+ serverProps
+ }
+
+ private def createReadCommittedConsumer(group: String) = {
+ val consumer = TestUtils.createConsumer(TestUtils.getBrokerListStrFromServers(servers),
+ groupId = group,
+ enableAutoCommit = false,
+ readCommitted = true)
+ transactionalConsumers += consumer
+ consumer
+ }
+
+ private def createTransactionalProducer(transactionalId: String): KafkaProducer[Array[Byte], Array[Byte]] = {
+ val producer = TestUtils.createTransactionalProducer(transactionalId, servers, maxInFlight = 1)
+ transactionalProducers += producer
+ producer
+ }
+}
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index eb4f80e..af6bbc4 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -1427,7 +1427,8 @@ object TestUtils extends Logging {
transactionTimeoutMs: Long = 60000,
maxBlockMs: Long = 60000,
deliveryTimeoutMs: Int = 120000,
- requestTimeoutMs: Int = 30000): KafkaProducer[Array[Byte], Array[Byte]] = {
+ requestTimeoutMs: Int = 30000,
+ maxInFlight: Int = 5): KafkaProducer[Array[Byte], Array[Byte]] = {
val props = new Properties()
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, TestUtils.getBrokerListStrFromServers(servers))
props.put(ProducerConfig.ACKS_CONFIG, "all")
@@ -1438,6 +1439,7 @@ object TestUtils extends Logging {
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs.toString)
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs.toString)
props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs.toString)
+ props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, maxInFlight.toString)
new KafkaProducer[Array[Byte], Array[Byte]](props, new ByteArraySerializer, new ByteArraySerializer)
}