From 254c077906029c31528c4ec76e1ad7ad28532905 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Thu, 26 Jul 2018 09:17:17 -0700 Subject: [PATCH 1/8] MINOR: MirrorMaker should set producer delivery timeout --- core/src/main/scala/kafka/tools/MirrorMaker.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala index 9cc6ebe1c86a8..6347f8ec50c6c 100755 --- a/core/src/main/scala/kafka/tools/MirrorMaker.scala +++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala @@ -193,7 +193,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { val sync = producerProps.getProperty("producer.type", "async").equals("sync") producerProps.remove("producer.type") // Defaults to no data loss settings. - maybeSetDefaultProperty(producerProps, ProducerConfig.RETRIES_CONFIG, Int.MaxValue.toString) + maybeSetDefaultProperty(producerProps, ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Int.MaxValue.toString) maybeSetDefaultProperty(producerProps, ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.MaxValue.toString) maybeSetDefaultProperty(producerProps, ProducerConfig.ACKS_CONFIG, "all") maybeSetDefaultProperty(producerProps, ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1") From f7494dd78535f2129d736c686a2f2dd1ac5fa0a6 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Thu, 26 Jul 2018 10:31:19 -0700 Subject: [PATCH 2/8] A few additional updates --- .../org/apache/kafka/clients/producer/KafkaProducer.java | 2 +- core/src/main/scala/kafka/tools/MirrorMaker.scala | 8 ++++---- docs/upgrade.html | 4 ++++ 3 files changed, 9 insertions(+), 5 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 b40b09a588a9c..2a35b30f259dc 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 @@ -96,7 +96,7 @@ * Properties props = new Properties(); * props.put("bootstrap.servers", "localhost:9092"); * props.put("acks", "all"); - * props.put("retries", 0); + * props.put("delivery.timeout.ms", 30000); * props.put("batch.size", 16384); * props.put("linger.ms", 1); * props.put("buffer.memory", 33554432); diff --git a/core/src/main/scala/kafka/tools/MirrorMaker.scala b/core/src/main/scala/kafka/tools/MirrorMaker.scala index 6347f8ec50c6c..d7e09e4efdb09 100755 --- a/core/src/main/scala/kafka/tools/MirrorMaker.scala +++ b/core/src/main/scala/kafka/tools/MirrorMaker.scala @@ -32,7 +32,7 @@ import org.apache.kafka.clients.consumer.{CommitFailedException, Consumer, Consu import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata} import org.apache.kafka.common.{KafkaException, TopicPartition} -import org.apache.kafka.common.serialization.ByteArrayDeserializer +import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer} import org.apache.kafka.common.utils.Utils import org.apache.kafka.common.errors.WakeupException import org.apache.kafka.common.record.RecordBatch @@ -50,7 +50,7 @@ import scala.util.control.ControlThrowable * @note For mirror maker, the following settings are set by default to make sure there is no data loss: * 1. use producer with following settings * acks=all - * retries=max integer + * delivery.timeout.ms=max integer * max.block.ms=max long * max.in.flight.requests.per.connection=1 * 2. Consumer Settings @@ -198,8 +198,8 @@ object MirrorMaker extends Logging with KafkaMetricsGroup { maybeSetDefaultProperty(producerProps, ProducerConfig.ACKS_CONFIG, "all") maybeSetDefaultProperty(producerProps, ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1") // Always set producer key and value serializer to ByteArraySerializer. - producerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") - producerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer") + producerProps.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[ByteArraySerializer].getName) + producerProps.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[ByteArraySerializer].getName) producer = new MirrorMakerProducer(sync, producerProps) // Create consumers diff --git a/docs/upgrade.html b/docs/upgrade.html index ac1388eb44005..264e26d6b0288 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -35,6 +35,10 @@

Upgrading from 0.8.x, 0.9.x, 0.1 which sets an upper bound on the total time between sending a record and receiving acknowledgement from the broker. By default, the delivery timeout is set to 2 minutes. +
  • By default, MirrorMaker now overrides delivery.timeout.ms to Integer.MAX_VALUE when + configuring the producer. If you have overridden the value of retries in order to fail faster, + you will instead need to override delivery.timeout.ms. +
    • From 81ebf295b730a9849910e7ceac8a7698fc91c703 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Thu, 26 Jul 2018 23:41:41 -0700 Subject: [PATCH 3/8] Prefer use of delivery timeout over retries --- .../clients/producer/ProducerConfig.java | 21 +++++++++------- .../apache/kafka/connect/runtime/Worker.java | 1 - .../storage/KafkaConfigBackingStore.java | 2 +- .../storage/KafkaOffsetBackingStore.java | 2 +- .../storage/KafkaStatusBackingStore.java | 2 +- .../kafka/api/AuthorizerIntegrationTest.scala | 2 -- .../kafka/api/BaseProducerSendTest.scala | 6 +++-- .../kafka/api/EndToEndAuthorizationTest.scala | 10 ++++---- .../kafka/api/IntegrationTestHarness.scala | 20 ++++++++-------- .../kafka/api/PlaintextConsumerTest.scala | 2 +- .../kafka/api/PlaintextProducerSendTest.scala | 3 +-- .../api/RackAwareAutoTopicCreationTest.scala | 2 +- ...aslClientsWithInvalidCredentialsTest.scala | 1 - .../DynamicBrokerReconfigurationTest.scala | 13 +++++----- .../kafka/ReplicationQuotasTestRig.scala | 2 +- ...FetchRequestDownConversionConfigTest.scala | 2 +- .../unit/kafka/server/FetchRequestTest.scala | 6 ++--- .../unit/kafka/server/LogRecoveryTest.scala | 1 - .../unit/kafka/server/ReplicaFetchTest.scala | 1 - .../kafka/server/ReplicationQuotasTest.scala | 8 +++---- .../kafka/server/ServerShutdownTest.scala | 1 - ...venReplicationProtocolAcceptanceTest.scala | 12 +++++----- .../epoch/LeaderEpochIntegrationTest.scala | 7 +++--- .../scala/unit/kafka/utils/TestUtils.scala | 15 ++++++------ .../log4jappender/KafkaLog4jAppender.java | 24 ++++++++++++++----- .../apache/kafka/streams/StreamsConfig.java | 11 +++++---- .../kafka/streams/StreamsConfigTest.java | 14 +++++------ .../QueryableStateIntegrationTest.java | 1 - .../kafka/streams/tests/SmokeTestClient.java | 2 +- .../kafka/streams/tests/SmokeTestDriver.java | 2 +- 30 files changed, 103 insertions(+), 93 deletions(-) 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 ab55353a66746..21e309f0c0b8f 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 @@ -107,10 +107,11 @@ public class ProducerConfig extends AbstractConfig { /** 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. " + DELIVERY_TIMEOUT_MS_CONFIG + "should be equal to or " - + "greater than " + REQUEST_TIMEOUT_MS_CONFIG + " + " + LINGER_MS_CONFIG; + private static final String DELIVERY_TIMEOUT_MS_DOC = "An upper bound on the time to report success or failure after " + + "Producer.send() returns. The producer may report failure to send a record earlier than this config if all " + + "the retries have been exhausted or a record is added to a batch nearing expiration. " + + "The value of this config should be equal to or greater than " + REQUEST_TIMEOUT_MS_CONFIG + + " + " + LINGER_MS_CONFIG + ". "; /** client.id */ public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG; @@ -181,10 +182,14 @@ public class ProducerConfig extends AbstractConfig { /** retries */ public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG; private static final String RETRIES_DOC = "Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error." - + " Note that this retry is no different than if the client resent the record upon receiving the error." - + " Allowing retries without setting " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to 1 will potentially change the" - + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second" - + " succeeds, then the records in the second batch may appear first."; + + " Note that this retry is no different than if the client resent the record upon receiving the error." + + " Allowing retries without setting " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to 1 will potentially change the" + + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second" + + " succeeds, then the records in the second batch may appear first. Note that the produce requests will be" + + " failed before the number of retries has been exhausted if the timeout configured by" + + " " + DELIVERY_TIMEOUT_MS_CONFIG + " expires first before successful acknowledgement. Users should generally" + + " prefer to leave this config unset and instead use " + DELIVERY_TIMEOUT_MS_CONFIG + " to control" + + " retry behavior."; /** key.serializer */ public static final String KEY_SERIALIZER_CLASS_CONFIG = "key.serializer"; 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 1f62103213c0e..e2fe6b607d734 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 @@ -135,7 +135,6 @@ public Worker( // These settings are designed to ensure there is no data loss. They *may* be overridden via configs passed to the // worker, but this may compromise the delivery guarantees of Kafka Connect. producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE)); - producerProps.put(ProducerConfig.RETRIES_CONFIG, Integer.toString(Integer.MAX_VALUE)); 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"); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java index ea196650c5e1b..e7ee632638d2e 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java @@ -415,7 +415,7 @@ KafkaBasedLog setupAndCreateKafkaBasedLog(String topic, final Wo Map producerProps = new HashMap<>(originals); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerProps.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); Map consumerProps = new HashMap<>(originals); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java index fb8ad97b48ddb..195c498edb76f 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java @@ -71,7 +71,7 @@ public void configure(final WorkerConfig config) { Map producerProps = new HashMap<>(originals); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerProps.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); Map consumerProps = new HashMap<>(originals); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName()); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java index 6710808f9a97a..6c829a090b944 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java @@ -128,7 +128,7 @@ public void configure(final WorkerConfig config) { Map producerProps = new HashMap<>(originals); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerProps.put(ProducerConfig.RETRIES_CONFIG, 0); // we handle retries in this class + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 0); // we handle retries in this class Map consumerProps = new HashMap<>(originals); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index c6b3af2d23095..dc9ca8568eab6 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -1442,7 +1442,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest { val transactionalProperties = new Properties() transactionalProperties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId) val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = 3, props = Some(transactionalProperties)) producers += producer producer @@ -1452,7 +1451,6 @@ class AuthorizerIntegrationTest extends BaseRequestTest { val idempotentProperties = new Properties() idempotentProperties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = 3, props = Some(idempotentProperties)) producers += producer producer diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala index 739675ead6e49..ad444259f3e67 100644 --- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala @@ -68,9 +68,11 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness { super.tearDown() } - protected def createProducer(brokerList: String, retries: Int = 0, lingerMs: Int = 0, props: Option[Properties] = None): KafkaProducer[Array[Byte],Array[Byte]] = { + protected def createProducer(brokerList: String, + lingerMs: Int = 0, + props: Option[Properties] = None): KafkaProducer[Array[Byte],Array[Byte]] = { val producer = TestUtils.createProducer(brokerList, securityProtocol = securityProtocol, trustStoreFile = trustStoreFile, - saslProperties = clientSaslProperties, retries = retries, lingerMs = lingerMs, props = props) + saslProperties = clientSaslProperties, lingerMs = lingerMs, props = props) registerProducer(producer) } diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala index a0023091bb02f..7ea761fae858e 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala @@ -189,11 +189,11 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas override def createProducer: KafkaProducer[Array[Byte], Array[Byte]] = { TestUtils.createProducer(brokerList, - maxBlockMs = 3000L, - securityProtocol = this.securityProtocol, - trustStoreFile = this.trustStoreFile, - saslProperties = this.clientSaslProperties, - props = Some(producerConfig)) + maxBlockMs = 3000L, + securityProtocol = this.securityProtocol, + trustStoreFile = this.trustStoreFile, + saslProperties = this.clientSaslProperties, + props = Some(producerConfig)) } /** diff --git a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala index 053f04e0cdc56..4601417f60615 100644 --- a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala +++ b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala @@ -93,19 +93,19 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness { } def createProducer: KafkaProducer[Array[Byte], Array[Byte]] = { - TestUtils.createProducer(brokerList, - securityProtocol = this.securityProtocol, - trustStoreFile = this.trustStoreFile, - saslProperties = this.clientSaslProperties, - props = Some(producerConfig)) + TestUtils.createProducer(brokerList, + securityProtocol = this.securityProtocol, + trustStoreFile = this.trustStoreFile, + saslProperties = this.clientSaslProperties, + props = Some(producerConfig)) } def createConsumer: KafkaConsumer[Array[Byte], Array[Byte]] = { - TestUtils.createConsumer(brokerList, - securityProtocol = this.securityProtocol, - trustStoreFile = this.trustStoreFile, - saslProperties = this.clientSaslProperties, - props = Some(consumerConfig)) + TestUtils.createConsumer(brokerList, + securityProtocol = this.securityProtocol, + trustStoreFile = this.trustStoreFile, + saslProperties = this.clientSaslProperties, + props = Some(consumerConfig)) } @After diff --git a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala index ba4df7d64f65b..0c8c771e0781b 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala @@ -619,7 +619,7 @@ class PlaintextConsumerTest extends BaseConsumerTest { producerProps.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.GZIP.name) producerProps.setProperty(ProducerConfig.LINGER_MS_CONFIG, Int.MaxValue.toString) val producer = TestUtils.createProducer(brokerList, securityProtocol = securityProtocol, trustStoreFile = trustStoreFile, - saslProperties = clientSaslProperties, retries = 0, lingerMs = Int.MaxValue, props = Some(producerProps)) + saslProperties = clientSaslProperties, lingerMs = Int.MaxValue, props = Some(producerProps)) (0 until numRecords).foreach { i => producer.send(new ProducerRecord(tp.topic, tp.partition, i.toLong, s"key $i".getBytes, s"value $i".getBytes)) } diff --git a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala index 1da6f9ee3694d..8ae39528b9884 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextProducerSendTest.scala @@ -70,8 +70,7 @@ class PlaintextProducerSendTest extends BaseProducerSendTest { */ @Test def testAutoCreateTopic() { - val producer = createProducer(brokerList, retries = 5) - + val producer = createProducer(brokerList) try { // Send a message to auto-create the topic val record = new ProducerRecord(topic, null, "key".getBytes, "value".getBytes) diff --git a/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala b/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala index bfd2924307d5c..5fc626bb0bbe5 100644 --- a/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala +++ b/core/src/test/scala/integration/kafka/api/RackAwareAutoTopicCreationTest.scala @@ -44,7 +44,7 @@ class RackAwareAutoTopicCreationTest extends KafkaServerTestHarness with RackAwa @Test def testAutoCreateTopic() { - val producer = TestUtils.createProducer(brokerList, retries = 5) + val producer = TestUtils.createProducer(brokerList) try { // Send a message to auto-create the topic val record = new ProducerRecord(topic, null, "key".getBytes, "value".getBytes) diff --git a/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala index b58ba74114020..ebc587e0d2fcb 100644 --- a/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala +++ b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala @@ -246,7 +246,6 @@ class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with val txProducer = TestUtils.createProducer(brokerList, securityProtocol = this.securityProtocol, saslProperties = this.clientSaslProperties, - retries = 1000, acks = -1, props = Some(producerConfig)) producers += txProducer diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 61d59192154c5..b75f69ba23184 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -304,7 +304,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet @Test def testLogCleanerConfig(): Unit = { - val (producerThread, consumerThread) = startProduceConsume(0) + val (producerThread, consumerThread) = startProduceConsume(retries = 0) verifyThreads("kafka-log-cleaner-thread-", countPerBroker = 1) @@ -437,7 +437,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet def testUncleanLeaderElectionEnable(): Unit = { val topic = "testtopic2" TestUtils.createTopic(zkClient, topic, 1, replicationFactor = 2, servers) - val producer = ProducerBuilder().maxRetries(1000).acks(1).build() + val producer = ProducerBuilder().acks(1).build() val consumer = ConsumerBuilder("unclean-leader-test").enableAutoCommit(false).topic(topic).build() verifyProduceConsume(producer, consumer, numRecords = 10, topic) consumer.commitSync() @@ -543,7 +543,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet def verifyThreadPoolResize(propName: String, currentSize: => Int, threadPrefix: String, mayReceiveDuplicates: Boolean): Unit = { maybeVerifyThreadPoolSize(propName, currentSize, threadPrefix) val numRetries = if (mayReceiveDuplicates) 100 else 0 - val (producerThread, consumerThread) = startProduceConsume(numRetries) + val (producerThread, consumerThread) = startProduceConsume(retries = numRetries) var threadPoolSize = currentSize (1 to 2).foreach { _ => threadPoolSize = reducePoolSize(propName, threadPoolSize, threadPrefix) @@ -1366,7 +1366,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } private case class ProducerBuilder() extends ClientBuilder[KafkaProducer[String, String]] { - private var _retries = 0 + private var _retries = Int.MaxValue private var _acks = -1 private var _requestTimeoutMs = 30000 @@ -1417,8 +1417,9 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet } } - private class ProducerThread(clientId: String, retries: Int) extends - ShutdownableThread(clientId, isInterruptible = false) { + private class ProducerThread(clientId: String, retries: Int) + extends ShutdownableThread(clientId, isInterruptible = false) { + private val producer = ProducerBuilder().maxRetries(retries).clientId(clientId).build() val lastSent = new ConcurrentHashMap[Int, Int]() @volatile var sent = 0 diff --git a/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala b/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala index 98e568bc3e42c..b2568c15d1886 100644 --- a/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala +++ b/core/src/test/scala/other/kafka/ReplicationQuotasTestRig.scala @@ -127,7 +127,7 @@ object ReplicationQuotasTestRig { createTopic(zkClient, topicName, replicas, servers) println("Writing Data") - val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), retries = 5, acks = 0) + val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), acks = 0) (0 until config.msgsPerPartition).foreach { x => (0 until config.partitions).foreach { partition => producer.send(new ProducerRecord(topicName, partition, null, new Array[Byte](config.msgSize))) diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestDownConversionConfigTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestDownConversionConfigTest.scala index 1e762b3f71620..1bf6f28eec515 100644 --- a/core/src/test/scala/unit/kafka/server/FetchRequestDownConversionConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/FetchRequestDownConversionConfigTest.scala @@ -52,7 +52,7 @@ class FetchRequestDownConversionConfigTest extends BaseRequestTest { private def initProducer(): Unit = { producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = 5, keySerializer = new StringSerializer, valueSerializer = new StringSerializer) + keySerializer = new StringSerializer, valueSerializer = new StringSerializer) } private def createTopics(numTopics: Int, numPartitions: Int, diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala index 57aca1e73ba84..8b4b924b28d04 100644 --- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala @@ -70,7 +70,7 @@ class FetchRequestTest extends BaseRequestTest { private def initProducer(): Unit = { producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = 5, keySerializer = new StringSerializer, valueSerializer = new StringSerializer) + keySerializer = new StringSerializer, valueSerializer = new StringSerializer) } @Test @@ -204,8 +204,8 @@ class FetchRequestTest extends BaseRequestTest { val propsOverride = new Properties propsOverride.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize.toString) val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = 5, lingerMs = Int.MaxValue, - keySerializer = new StringSerializer, valueSerializer = new ByteArraySerializer, props = Some(propsOverride)) + lingerMs = Int.MaxValue, keySerializer = new StringSerializer, + valueSerializer = new ByteArraySerializer, props = Some(propsOverride)) val bytes = new Array[Byte](msgValueLen) val futures = try { (0 to 1000).map { _ => diff --git a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala index 880950ae02c75..d65cced87abc1 100755 --- a/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogRecoveryTest.scala @@ -67,7 +67,6 @@ class LogRecoveryTest extends ZooKeeperTestHarness { producer.close() producer = TestUtils.createProducer( TestUtils.getBrokerListStrFromServers(servers), - retries = 5, keySerializer = new IntegerSerializer, valueSerializer = new StringSerializer ) diff --git a/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala index 0dd22f14f6eb9..8eba82471007e 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicaFetchTest.scala @@ -56,7 +56,6 @@ class ReplicaFetchTest extends ZooKeeperTestHarness { // send test messages to leader val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(brokers), - retries = 5, keySerializer = new StringSerializer, valueSerializer = new StringSerializer) val records = testMessageList1.map(m => new ProducerRecord(topic1, m, m)) ++ diff --git a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala index 0bbe637e22361..51254866ebe41 100644 --- a/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala +++ b/core/src/test/scala/unit/kafka/server/ReplicationQuotasTest.scala @@ -114,7 +114,7 @@ class ReplicationQuotasTest extends ZooKeeperTestHarness { adminZkClient.changeTopicConfig(topic, propsWith(FollowerReplicationThrottledReplicasProp, "0:106,1:106,2:106,3:107,4:107,5:107")) //Add data equally to each partition - producer = createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 1) + producer = createProducer(getBrokerListStrFromServers(brokers), acks = 1) (0 until msgCount).foreach { _ => (0 to 7).foreach { partition => producer.send(new ProducerRecord(topic, partition, null, msg)) @@ -203,14 +203,14 @@ class ReplicationQuotasTest extends ZooKeeperTestHarness { val throttledTook = System.currentTimeMillis() - start - assertTrue((s"Throttled replication of ${throttledTook}ms should be > ${expectedDuration * 1000 * 0.9}ms"), + assertTrue(s"Throttled replication of ${throttledTook}ms should be > ${expectedDuration * 1000 * 0.9}ms", throttledTook > expectedDuration * 1000 * 0.9) - assertTrue((s"Throttled replication of ${throttledTook}ms should be < ${expectedDuration * 1500}ms"), + assertTrue(s"Throttled replication of ${throttledTook}ms should be < ${expectedDuration * 1500}ms", throttledTook < expectedDuration * 1000 * 1.5) } def addData(msgCount: Int, msg: Array[Byte]): Unit = { - producer = createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 0) + producer = createProducer(getBrokerListStrFromServers(brokers), acks = 0) (0 until msgCount).map(_ => producer.send(new ProducerRecord(topic, msg))).foreach(_.get) waitForOffsetsToMatch(msgCount, 0, 100) } diff --git a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala index 9f966b448b70e..ff097496e02ed 100755 --- a/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala +++ b/core/src/test/scala/unit/kafka/server/ServerShutdownTest.scala @@ -54,7 +54,6 @@ class ServerShutdownTest extends ZooKeeperTestHarness { def createProducer(server: KafkaServer): KafkaProducer[Integer, String] = TestUtils.createProducer( TestUtils.getBrokerListStrFromServers(Seq(server)), - retries = 5, keySerializer = new IntegerSerializer, valueSerializer = new StringSerializer ) diff --git a/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala b/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala index 3dcf4ff74f789..149f05f0c104c 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/EpochDrivenReplicationProtocolAcceptanceTest.scala @@ -305,7 +305,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness TestUtils.createTopic(zkClient, topic, Map(0 -> Seq(100, 101)), brokers, CoreUtils.propsWith((KafkaConfig.MinInSyncReplicasProp, "1"))) - producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 1) + producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = 1) // Write one message while both brokers are up (0 until 1).foreach { i => @@ -328,7 +328,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness //Bounce the producer (this is required, probably because the broker port changes on restart?) producer.close() - producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 1) + producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = 1) //Write 3 messages (0 until 3).foreach { i => @@ -340,7 +340,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness //Bounce the producer (this is required, probably because the broker port changes on restart?) producer.close() - producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 1) + producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = 1) //Write 1 message (0 until 1).foreach { i => @@ -352,7 +352,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness //Bounce the producer (this is required, probably because the broker port changes on restart?) producer.close() - producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = 1) + producer = TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = 1) //Write 2 messages (0 until 2).foreach { i => @@ -411,7 +411,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness } private def createBufferingProducer: KafkaProducer[Array[Byte], Array[Byte]] = { - TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = -1, lingerMs = 10000, + TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = -1, lingerMs = 10000, props = Option(CoreUtils.propsWith( (ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(msg.length * 1000)) , (ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy") @@ -451,7 +451,7 @@ class EpochDrivenReplicationProtocolAcceptanceTest extends ZooKeeperTestHarness } private def createProducer: KafkaProducer[Array[Byte], Array[Byte]] = { - TestUtils.createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = -1) + TestUtils.createProducer(getBrokerListStrFromServers(brokers), acks = -1) } private def leader(): KafkaServer = { diff --git a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala index 4b5a0928a483a..a6b7732581114 100644 --- a/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/server/epoch/LeaderEpochIntegrationTest.scala @@ -104,7 +104,7 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging { TestUtils.createTopic(zkClient, topic2, assignment2, brokers) //Send messages equally to the two partitions, then half as many to a third - producer = createProducer(getBrokerListStrFromServers(brokers), retries = 5, acks = -1) + producer = createProducer(getBrokerListStrFromServers(brokers), acks = -1) (0 until 10).foreach { _ => producer.send(new ProducerRecord(topic1, 0, null, "IHeartLogs".getBytes)) } @@ -144,7 +144,7 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging { brokers = Seq(100, 101).map { id => createServer(fromProps(createBrokerConfig(id, zkConnect))) } def leo() = brokers(1).replicaManager.getReplica(tp).get.logEndOffset.messageOffset TestUtils.createTopic(zkClient, tp.topic, Map(tp.partition -> Seq(101)), brokers) - producer = createProducer(getBrokerListStrFromServers(brokers), retries = 10, acks = -1) + producer = createProducer(getBrokerListStrFromServers(brokers), acks = -1) //1. Given a single message producer.send(new ProducerRecord(tp.topic, tp.partition, null, "IHeartLogs".getBytes)).get @@ -251,7 +251,8 @@ class LeaderEpochIntegrationTest extends ZooKeeperTestHarness with Logging { private def sendFourMessagesToEachTopic() = { val testMessageList1 = List("test1", "test2", "test3", "test4") val testMessageList2 = List("test5", "test6", "test7", "test8") - val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(brokers), retries = 5, keySerializer = new StringSerializer, valueSerializer = new StringSerializer) + val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(brokers), + keySerializer = new StringSerializer, valueSerializer = new StringSerializer) val records = testMessageList1.map(m => new ProducerRecord(topic1, m, m)) ++ testMessageList2.map(m => new ProducerRecord(topic2, m, m)) diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index aa902f2582b7c..ea50a703c7195 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -547,7 +547,8 @@ object TestUtils extends Logging { acks: Int = -1, maxBlockMs: Long = 60 * 1000L, bufferSize: Long = 1024L * 1024L, - retries: Int = 0, + retries: Int = Int.MaxValue, + deliveryTimeoutMs: Int = 20000, lingerMs: Int = 0, requestTimeoutMs: Int = 30 * 1000, securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT, @@ -993,7 +994,7 @@ object TestUtils extends Logging { compressionType: CompressionType = CompressionType.NONE): Unit = { val props = new Properties() props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType.name) - val producer = createProducer(TestUtils.getBrokerListStrFromServers(servers), retries = 5, acks = acks) + val producer = createProducer(TestUtils.getBrokerListStrFromServers(servers), acks = acks) try { val futures = records.map(producer.send) futures.foreach(_.get) @@ -1017,10 +1018,7 @@ object TestUtils extends Logging { } def produceMessage(servers: Seq[KafkaServer], topic: String, message: String) { - val producer = createProducer( - TestUtils.getBrokerListStrFromServers(servers), - retries = 5 - ) + val producer = createProducer(TestUtils.getBrokerListStrFromServers(servers)) producer.send(new ProducerRecord(topic, topic.getBytes, message.getBytes)).get producer.close() } @@ -1277,7 +1275,8 @@ object TestUtils extends Logging { props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize.toString) props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, transactionTimeoutMs.toString) - TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), retries = Integer.MAX_VALUE, acks = -1, props = Some(props)) + TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), + deliveryTimeoutMs = Int.MaxValue, acks = -1, props = Some(props)) } // Seeds the given topic with records with keys and values in the range [0..numRecords) @@ -1285,7 +1284,7 @@ object TestUtils extends Logging { val props = new Properties() props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - retries = Integer.MAX_VALUE, acks = -1, props = Some(props)) + deliveryTimeoutMs = Int.MaxValue, acks = -1, props = Some(props)) try { for (i <- 0 until numRecords) { producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, asBytes(i.toString), asBytes(i.toString))) diff --git a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java index 6a09cab3f3b7d..00d9c8b188c95 100644 --- a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java +++ b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java @@ -34,6 +34,7 @@ import static org.apache.kafka.clients.producer.ProducerConfig.COMPRESSION_TYPE_CONFIG; import static org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG; +import static org.apache.kafka.clients.producer.ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG; import static org.apache.kafka.clients.producer.ProducerConfig.RETRIES_CONFIG; import static org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG; import static org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG; @@ -64,8 +65,9 @@ public class KafkaLog4jAppender extends AppenderSkeleton { private String clientJaasConfPath; private String kerb5ConfPath; - private int retries; - private int requiredNumAcks = Integer.MAX_VALUE; + private Integer retries; + private Integer requiredNumAcks; + private Integer deliveryTimeoutMs; private boolean syncSend; private Producer producer; @@ -81,7 +83,7 @@ public void setBrokerList(String brokerList) { this.brokerList = brokerList; } - public int getRequiredNumAcks() { + public Integer getRequiredNumAcks() { return requiredNumAcks; } @@ -89,7 +91,7 @@ public void setRequiredNumAcks(int requiredNumAcks) { this.requiredNumAcks = requiredNumAcks; } - public int getRetries() { + public Integer getRetries() { return retries; } @@ -97,6 +99,14 @@ public void setRetries(int retries) { this.retries = retries; } + public Integer getDeliveryTimeoutMs() { + return deliveryTimeoutMs; + } + + public void setDeliveryTimeoutMs(int deliveryTimeoutMs) { + this.deliveryTimeoutMs = deliveryTimeoutMs; + } + public String getCompressionType() { return compressionType; } @@ -205,10 +215,12 @@ public void activateOptions() { throw new ConfigException("Topic must be specified by the Kafka log4j appender"); if (compressionType != null) props.put(COMPRESSION_TYPE_CONFIG, compressionType); - if (requiredNumAcks != Integer.MAX_VALUE) + if (requiredNumAcks != null) props.put(ACKS_CONFIG, Integer.toString(requiredNumAcks)); - if (retries > 0) + if (retries != null) props.put(RETRIES_CONFIG, retries); + if (deliveryTimeoutMs != null) + props.put(DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs); if (securityProtocol != null) { props.put(SECURITY_PROTOCOL_CONFIG, securityProtocol); } diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java index 54fcbc0bb0a59..ac5b8fd2543a0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java @@ -98,10 +98,11 @@ * StreamsConfig streamsConfig = new StreamsConfig(streamsProperties); * } * - * When increasing both {@link ProducerConfig#RETRIES_CONFIG} and {@link ProducerConfig#MAX_BLOCK_MS_CONFIG} to be more resilient to non-available brokers you should also - * consider increasing {@link ConsumerConfig#MAX_POLL_INTERVAL_MS_CONFIG} using the following guidance: + * + * When increasing {@link ProducerConfig#MAX_BLOCK_MS_CONFIG} to be more resilient to non-available brokers you should also + * increase {@link ConsumerConfig#MAX_POLL_INTERVAL_MS_CONFIG} using the following guidance: * 
    - *     max.poll.interval.ms > min ( max.block.ms, (retries +1) * request.timeout.ms )
+ *     max.poll.interval.ms > max.block.ms
  *
    * * @@ -687,7 +688,7 @@ public class StreamsConfig extends AbstractConfig { static { final Map tempProducerDefaultOverrides = new HashMap<>(); tempProducerDefaultOverrides.put(ProducerConfig.LINGER_MS_CONFIG, "100"); - tempProducerDefaultOverrides.put(ProducerConfig.RETRIES_CONFIG, 10); + tempProducerDefaultOverrides.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); PRODUCER_DEFAULT_OVERRIDES = Collections.unmodifiableMap(tempProducerDefaultOverrides); } @@ -695,7 +696,7 @@ public class StreamsConfig extends AbstractConfig { private static final Map PRODUCER_EOS_OVERRIDES; static { final Map tempProducerDefaultOverrides = new HashMap<>(PRODUCER_DEFAULT_OVERRIDES); - tempProducerDefaultOverrides.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); + tempProducerDefaultOverrides.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); tempProducerDefaultOverrides.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true); PRODUCER_EOS_OVERRIDES = Collections.unmodifiableMap(tempProducerDefaultOverrides); diff --git a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java index cdd4d097ba377..9755334d09fb6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java @@ -456,9 +456,7 @@ public void shouldResetToDefaultIfConsumerIsolationLevelIsOverriddenIfEosEnabled props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "anyValue"); final StreamsConfig streamsConfig = new StreamsConfig(props); final Map consumerConfigs = streamsConfig.getMainConsumerConfigs("groupId", "clientId"); - String isoLevel = (String) consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG); - String name = READ_COMMITTED.name(); - assertThat((String) consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_COMMITTED.name().toLowerCase(Locale.ROOT))); + assertThat(consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_COMMITTED.name().toLowerCase(Locale.ROOT))); } @Test @@ -466,7 +464,7 @@ public void shouldAllowSettingConsumerIsolationLevelIfEosDisabled() { props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT)); final StreamsConfig streamsConfig = new StreamsConfig(props); final Map consumerConfigs = streamsConfig.getMainConsumerConfigs("groupId", "clientrId"); - assertThat((String) consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT))); + assertThat(consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_UNCOMMITTED.name().toLowerCase(Locale.ROOT))); } @@ -484,7 +482,7 @@ public void shouldAllowSettingProducerEnableIdempotenceIfEosDisabled() { props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false); final StreamsConfig streamsConfig = new StreamsConfig(props); final Map producerConfigs = streamsConfig.getProducerConfigs("clientId"); - assertThat((Boolean) producerConfigs.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), equalTo(false)); + assertThat(producerConfigs.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), equalTo(false)); } @Test @@ -495,9 +493,9 @@ public void shouldSetDifferentDefaultsIfEosEnabled() { final Map consumerConfigs = streamsConfig.getMainConsumerConfigs("groupId", "clientId"); final Map producerConfigs = streamsConfig.getProducerConfigs("clientId"); - assertThat((String) consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_COMMITTED.name().toLowerCase(Locale.ROOT))); + assertThat(consumerConfigs.get(ConsumerConfig.ISOLATION_LEVEL_CONFIG), equalTo(READ_COMMITTED.name().toLowerCase(Locale.ROOT))); assertTrue((Boolean) producerConfigs.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)); - assertThat((Integer) producerConfigs.get(ProducerConfig.RETRIES_CONFIG), equalTo(Integer.MAX_VALUE)); + assertThat(producerConfigs.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG), equalTo(Integer.MAX_VALUE)); assertThat(streamsConfig.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG), equalTo(100L)); } @@ -510,7 +508,7 @@ public void shouldNotOverrideUserConfigRetriesIfExactlyOnceEnabled() { final Map producerConfigs = streamsConfig.getProducerConfigs("clientId"); - assertThat((Integer) producerConfigs.get(ProducerConfig.RETRIES_CONFIG), equalTo(numberOfRetries)); + assertThat(producerConfigs.get(ProducerConfig.RETRIES_CONFIG), equalTo(numberOfRetries)); } @Test diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java index ff791be131691..496ba5842a9c6 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java @@ -1065,7 +1065,6 @@ public void run() { final Properties producerConfig = new Properties(); producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); producerConfig.put(ProducerConfig.ACKS_CONFIG, "all"); - producerConfig.put(ProducerConfig.RETRIES_CONFIG, 10); producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java index 268dd2fce21f6..9a6849c965578 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java @@ -105,7 +105,7 @@ private static Properties getStreamsConfig(final Properties props) { fullProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); fullProps.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3); fullProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - fullProps.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); + fullProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); fullProps.put(ProducerConfig.ACKS_CONFIG, "all"); //TODO remove this config or set to smaller value when KIP-91 is merged diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java index 7533fdd085883..f3b568e265dc8 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java @@ -154,7 +154,7 @@ public static Map> generate(final String kafka, producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); // the next 2 config values make sure that all records are produced with no loss and no duplicates - producerProps.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 80000); From 7ed56b21145a414209d197f192e1ad988abd173d Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 27 Jul 2018 09:01:45 -0700 Subject: [PATCH 4/8] A few more cleanups and review comments --- .../connect/storage/KafkaStatusBackingStore.java | 2 +- .../test/scala/unit/kafka/utils/TestUtils.scala | 16 +++++----------- .../kafka/streams/perf/SimpleBenchmark.java | 3 --- .../kafka/streams/perf/YahooBenchmark.java | 2 -- .../streams/tests/BrokerCompatibilityTest.java | 2 -- .../kafka/streams/tests/EosTestClient.java | 2 -- .../kafka/streams/tests/SmokeTestClient.java | 4 ---- .../kafka/streams/tests/SmokeTestDriver.java | 5 +---- .../kafka/tools/TransactionalMessageCopier.java | 6 ------ 9 files changed, 7 insertions(+), 35 deletions(-) diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java index 6c829a090b944..6710808f9a97a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java @@ -128,7 +128,7 @@ public void configure(final WorkerConfig config) { Map producerProps = new HashMap<>(originals); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); - producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 0); // we handle retries in this class + producerProps.put(ProducerConfig.RETRIES_CONFIG, 0); // we handle retries in this class Map consumerProps = new HashMap<>(originals); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index ea50a703c7195..978362782ac39 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -548,9 +548,9 @@ object TestUtils extends Logging { maxBlockMs: Long = 60 * 1000L, bufferSize: Long = 1024L * 1024L, retries: Int = Int.MaxValue, - deliveryTimeoutMs: Int = 20000, + deliveryTimeoutMs: Int = 30 * 1000, lingerMs: Int = 0, - requestTimeoutMs: Int = 30 * 1000, + requestTimeoutMs: Int = 20 * 1000, securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT, trustStoreFile: Option[File] = None, saslProperties: Option[Properties] = None, @@ -564,13 +564,10 @@ object TestUtils extends Logging { producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs.toString) producerProps.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferSize.toString) producerProps.put(ProducerConfig.RETRIES_CONFIG, retries.toString) + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs.toString) producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs.toString) producerProps.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs.toString) - // In case of overflow set maximum possible value for deliveryTimeoutMs - val deliveryTimeoutMs = if (lingerMs + requestTimeoutMs < 0) Int.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", @@ -1271,20 +1268,17 @@ object TestUtils extends Logging { transactionTimeoutMs: Long = 60000) = { val props = new Properties() props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId) - props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5") props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize.toString) props.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, transactionTimeoutMs.toString) - TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - deliveryTimeoutMs = Int.MaxValue, acks = -1, props = Some(props)) + TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), acks = -1, props = Some(props)) } // Seeds the given topic with records with keys and values in the range [0..numRecords) def seedTopicWithNumberedRecords(topic: String, numRecords: Int, servers: Seq[KafkaServer]): Unit = { val props = new Properties() props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") - val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), - deliveryTimeoutMs = Int.MaxValue, acks = -1, props = Some(props)) + val producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), acks = -1, props = Some(props)) try { for (i <- 0 until numRecords) { producer.send(new ProducerRecord[Array[Byte], Array[Byte]](topic, asBytes(i.toString), asBytes(i.toString))) diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java index 654fd03f10ab2..9301e5f102949 100644 --- a/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java +++ b/streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java @@ -262,9 +262,6 @@ public void setStreamProperties(final String applicationId) { // improve producer throughput props.put(ProducerConfig.LINGER_MS_CONFIG, 5000); props.put(ProducerConfig.BATCH_SIZE_CONFIG, 128 * 1024); - - //TODO remove this config or set to smaller value when KIP-91 is merged - props.put(StreamsConfig.producerPrefix(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), 60000); } private Properties setProduceConsumeProperties(final String clientId) { diff --git a/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java b/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java index ae538703dd3b7..21042212ce1a8 100644 --- a/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java +++ b/streams/src/test/java/org/apache/kafka/streams/perf/YahooBenchmark.java @@ -177,8 +177,6 @@ public void run() { final CountDownLatch latch = new CountDownLatch(1); parent.setStreamProperties("simple-benchmark-yahoo" + new Random().nextInt()); - //TODO remove this config or set to smaller value when KIP-91 is merged - parent.props.put(StreamsConfig.producerPrefix(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), 60000); final KafkaStreams streams = createYahooBenchmarkStreams(parent.props, campaignsTopic, eventsTopic, latch, parent.numRecords); parent.runGenericBenchmark(streams, "Streams Yahoo Performance [records/latency/rec-sec/MB-sec counted]: ", latch); diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/BrokerCompatibilityTest.java b/streams/src/test/java/org/apache/kafka/streams/tests/BrokerCompatibilityTest.java index 3c8446ca46652..767c9f16e4851 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/BrokerCompatibilityTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/BrokerCompatibilityTest.java @@ -80,8 +80,6 @@ public static void main(final String[] args) throws IOException { streamsProperties.put(StreamsConfig.consumerPrefix(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG), timeout); streamsProperties.put(StreamsConfig.consumerPrefix(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG), timeout); streamsProperties.put(StreamsConfig.REQUEST_TIMEOUT_MS_CONFIG, timeout + 1); - //TODO remove this config or set to smaller value when KIP-91 is merged - streamsProperties.put(StreamsConfig.producerPrefix(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), 60000); Serde stringSerde = Serdes.String(); diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java index 2d39d53836b96..784064a41833d 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java @@ -119,8 +119,6 @@ private KafkaStreams createKafkaStreams(final Properties props) { props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); - //TODO remove this config or set to smaller value when KIP-91 is merged - props.put(StreamsConfig.producerPrefix(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), 60000); final StreamsBuilder builder = new StreamsBuilder(); final KStream data = builder.stream("data"); diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java index 9a6849c965578..79dfb30ef2cb8 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestClient.java @@ -105,12 +105,8 @@ private static Properties getStreamsConfig(final Properties props) { fullProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); fullProps.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3); fullProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - fullProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); fullProps.put(ProducerConfig.ACKS_CONFIG, "all"); - //TODO remove this config or set to smaller value when KIP-91 is merged - fullProps.put(StreamsConfig.producerPrefix(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG), 80000); - fullProps.putAll(props); return fullProps; } diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java index f3b568e265dc8..a504333a7d1e8 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/SmokeTestDriver.java @@ -153,10 +153,7 @@ public static Map> generate(final String kafka, producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka); producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); - // the next 2 config values make sure that all records are produced with no loss and no duplicates - producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.MAX_VALUE); producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); - producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 80000); final KafkaProducer producer = new KafkaProducer<>(producerProps); @@ -166,7 +163,7 @@ public static Map> generate(final String kafka, final ValueList[] data = new ValueList[numKeys]; for (int i = 0; i < numKeys; i++) { data[i] = new ValueList(i, i + maxRecordsPerKey - 1); - allData.put(data[i].key, new HashSet()); + allData.put(data[i].key, new HashSet<>()); } final Random rand = new Random(); 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 87d27e8f2590b..0d74645379ebe 100644 --- a/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java +++ b/tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java @@ -155,12 +155,6 @@ private static KafkaProducer createProducer(Namespace parsedArgs props.put(ProducerConfig.BATCH_SIZE_CONFIG, "512"); props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); - // Multiple inflights means that when there are rolling bounces and other cluster instability, there is an - // increased likelihood of having previously tried batch expire in the accumulator. This is a fatal error - // for a transaction, causing the copier to exit. To work around this, we bump the request timeout. - // We can get rid of this when KIP-91 is merged. - props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000"); - return new KafkaProducer<>(props); } From 5a1f744d0d56c8790651d84a1277b0b8d8da9407 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 27 Jul 2018 09:59:41 -0700 Subject: [PATCH 5/8] Fix checkstyle error in EosTestClient --- .../test/java/org/apache/kafka/streams/tests/EosTestClient.java | 1 - 1 file changed, 1 deletion(-) diff --git a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java index 784064a41833d..6069298033d21 100644 --- a/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java +++ b/streams/src/test/java/org/apache/kafka/streams/tests/EosTestClient.java @@ -16,7 +16,6 @@ */ package org.apache.kafka.streams.tests; -import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.KafkaStreams; From e02c5ec2acf8491686c316e0522256dac8b4cb74 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Fri, 27 Jul 2018 11:52:10 -0700 Subject: [PATCH 6/8] Address review comments --- .../clients/producer/ProducerConfig.java | 15 +++++++----- .../log4jappender/KafkaLog4jAppender.java | 23 +++++++++---------- .../apache/kafka/streams/StreamsConfig.java | 2 -- 3 files changed, 20 insertions(+), 20 deletions(-) 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 21e309f0c0b8f..6b37e3cc9befc 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 @@ -107,11 +107,14 @@ public class ProducerConfig extends AbstractConfig { /** 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. The producer may report failure to send a record earlier than this config if all " - + "the retries have been exhausted or a record is added to a batch nearing expiration. " - + "The value of this config should be equal to or greater than " + REQUEST_TIMEOUT_MS_CONFIG - + " + " + LINGER_MS_CONFIG + ". "; + private static final String DELIVERY_TIMEOUT_MS_DOC = "An upper bound on the time to report success or failure " + + "after a call to send() returns. This limits the total time that a record will be delayed " + + "prior to sending, the time to await acknowledgement from the broker (if expected), and the time allowed " + + "for retriable send failures. The producer may report failure to send a record earlier than this config if " + + "either an unrecoverable error is encountered, the retries have been exhausted, " + + "or the record is added to a batch which reached an earlier delivery expiration deadline." + + "The value of this config should be greater than or equal to the sum of " + REQUEST_TIMEOUT_MS_CONFIG + + " and " + LINGER_MS_CONFIG + ". "; /** client.id */ public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG; @@ -185,7 +188,7 @@ public class ProducerConfig extends AbstractConfig { + " Note that this retry is no different than if the client resent the record upon receiving the error." + " Allowing retries without setting " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to 1 will potentially change the" + " ordering of records because if two batches are sent to a single partition, and the first fails and is retried but the second" - + " succeeds, then the records in the second batch may appear first. Note that the produce requests will be" + + " succeeds, then the records in the second batch may appear first. Note additionall that produce requests will be" + " failed before the number of retries has been exhausted if the timeout configured by" + " " + DELIVERY_TIMEOUT_MS_CONFIG + " expires first before successful acknowledgement. Users should generally" + " prefer to leave this config unset and instead use " + DELIVERY_TIMEOUT_MS_CONFIG + " to control" diff --git a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java index 00d9c8b188c95..6ddaf929a6d6f 100644 --- a/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java +++ b/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java @@ -65,9 +65,9 @@ public class KafkaLog4jAppender extends AppenderSkeleton { private String clientJaasConfPath; private String kerb5ConfPath; - private Integer retries; - private Integer requiredNumAcks; - private Integer deliveryTimeoutMs; + private int retries = Integer.MAX_VALUE; + private int requiredNumAcks = 1; + private int deliveryTimeoutMs = 120000; private boolean syncSend; private Producer producer; @@ -83,7 +83,7 @@ public void setBrokerList(String brokerList) { this.brokerList = brokerList; } - public Integer getRequiredNumAcks() { + public int getRequiredNumAcks() { return requiredNumAcks; } @@ -91,7 +91,7 @@ public void setRequiredNumAcks(int requiredNumAcks) { this.requiredNumAcks = requiredNumAcks; } - public Integer getRetries() { + public int getRetries() { return retries; } @@ -99,7 +99,7 @@ public void setRetries(int retries) { this.retries = retries; } - public Integer getDeliveryTimeoutMs() { + public int getDeliveryTimeoutMs() { return deliveryTimeoutMs; } @@ -215,12 +215,11 @@ public void activateOptions() { throw new ConfigException("Topic must be specified by the Kafka log4j appender"); if (compressionType != null) props.put(COMPRESSION_TYPE_CONFIG, compressionType); - if (requiredNumAcks != null) - props.put(ACKS_CONFIG, Integer.toString(requiredNumAcks)); - if (retries != null) - props.put(RETRIES_CONFIG, retries); - if (deliveryTimeoutMs != null) - props.put(DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs); + + props.put(ACKS_CONFIG, Integer.toString(requiredNumAcks)); + props.put(RETRIES_CONFIG, retries); + props.put(DELIVERY_TIMEOUT_MS_CONFIG, deliveryTimeoutMs); + if (securityProtocol != null) { props.put(SECURITY_PROTOCOL_CONFIG, securityProtocol); } diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java index ac5b8fd2543a0..e9fe3c4698421 100644 --- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java +++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java @@ -688,8 +688,6 @@ public class StreamsConfig extends AbstractConfig { static { final Map tempProducerDefaultOverrides = new HashMap<>(); tempProducerDefaultOverrides.put(ProducerConfig.LINGER_MS_CONFIG, "100"); - tempProducerDefaultOverrides.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120000); - PRODUCER_DEFAULT_OVERRIDES = Collections.unmodifiableMap(tempProducerDefaultOverrides); } From 2d3670b99a08bfc4696cb683b61931d76f4f9bf8 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Mon, 30 Jul 2018 13:03:56 -0700 Subject: [PATCH 7/8] Use retries=0 in test case to ensure expected error is raised --- .../integration/kafka/api/ProducerFailureHandlingTest.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala index 0227690e928f9..b7d3ecb686a97 100644 --- a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala +++ b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala @@ -64,11 +64,11 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness { override def setUp() { super.setUp() - producer1 = TestUtils.createProducer(brokerList, acks = 0, requestTimeoutMs = 30000, maxBlockMs = 10000L, + producer1 = TestUtils.createProducer(brokerList, acks = 0, retries = 0, requestTimeoutMs = 30000, maxBlockMs = 10000L, bufferSize = producerBufferSize) - producer2 = TestUtils.createProducer(brokerList, acks = 1, requestTimeoutMs = 30000, maxBlockMs = 10000L, + producer2 = TestUtils.createProducer(brokerList, acks = 1, retries = 0, requestTimeoutMs = 30000, maxBlockMs = 10000L, bufferSize = producerBufferSize) - producer3 = TestUtils.createProducer(brokerList, acks = -1, requestTimeoutMs = 30000, maxBlockMs = 10000L, + producer3 = TestUtils.createProducer(brokerList, acks = -1, retries = 0, requestTimeoutMs = 30000, maxBlockMs = 10000L, bufferSize = producerBufferSize) } From eae84c1396d2828a0a2a4aa9fb571d5f7d39a652 Mon Sep 17 00:00:00 2001 From: Jason Gustafson Date: Tue, 31 Jul 2018 17:34:21 -0700 Subject: [PATCH 8/8] Fix failing tests --- .../DynamicBrokerReconfigurationTest.scala | 4 ++++ .../unit/kafka/server/FetchRequestTest.scala | 3 ++- .../unit/kafka/server/LogDirFailureTest.scala | 16 +++++++++++----- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index b75f69ba23184..5694cff2c9091 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -736,6 +736,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet val producer1 = ProducerBuilder().trustStoreProps(sslProperties1) .maxRetries(0) .requestTimeoutMs(1000) + .deliveryTimeoutMs(1000) .bootstrapServers(bootstrap) .build() @@ -1369,15 +1370,18 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet private var _retries = Int.MaxValue private var _acks = -1 private var _requestTimeoutMs = 30000 + private var _deliveryTimeoutMs = 30000 def maxRetries(retries: Int): ProducerBuilder = { _retries = retries; this } def acks(acks: Int): ProducerBuilder = { _acks = acks; this } def requestTimeoutMs(timeoutMs: Int): ProducerBuilder = { _requestTimeoutMs = timeoutMs; this } + def deliveryTimeoutMs(timeoutMs: Int): ProducerBuilder = { _deliveryTimeoutMs= timeoutMs; this } override def build(): KafkaProducer[String, String] = { val producer = TestUtils.createProducer(bootstrapServers, acks = _acks, requestTimeoutMs = _requestTimeoutMs, + deliveryTimeoutMs = _deliveryTimeoutMs, retries = _retries, securityProtocol = _securityProtocol, trustStoreFile = Some(trustStoreFile1), diff --git a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala index 8b4b924b28d04..388b0f85c6ed0 100644 --- a/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/FetchRequestTest.scala @@ -263,7 +263,8 @@ class FetchRequestTest extends BaseRequestTest { // Increase linger so that we have control over the batches created producer = TestUtils.createProducer(TestUtils.getBrokerListStrFromServers(servers), retries = 5, keySerializer = new StringSerializer, valueSerializer = new StringSerializer, - lingerMs = 300 * 1000) + lingerMs = 30 * 1000, + deliveryTimeoutMs = 60 * 1000) val topicConfig = Map(LogConfig.MessageFormatVersionProp -> KAFKA_0_11_0_IV2.version) val (topicPartition, leaderId) = createTopics(numTopics = 1, numPartitions = 1, topicConfig).head diff --git a/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala b/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala index 2087363bfbd7a..9b32cd26d9453 100644 --- a/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala @@ -24,9 +24,8 @@ import kafka.server.LogDirFailureTest._ import kafka.api.IntegrationTestHarness import kafka.controller.{OfflineReplica, PartitionAndReplica} import kafka.utils.{CoreUtils, Exit, TestUtils} - import org.apache.kafka.clients.consumer.KafkaConsumer -import org.apache.kafka.clients.producer.{ProducerConfig, ProducerRecord} +import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} import org.apache.kafka.common.TopicPartition import org.apache.kafka.common.utils.Utils import org.apache.kafka.common.errors.{KafkaStorageException, NotLeaderForPartitionException} @@ -47,8 +46,6 @@ class LogDirFailureTest extends IntegrationTestHarness { private val partitionNum = 12 this.logDirCount = 3 - this.producerConfig.setProperty(ProducerConfig.RETRIES_CONFIG, "0") - this.producerConfig.setProperty(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, "100") this.serverConfig.setProperty(KafkaConfig.ReplicaHighWatermarkCheckpointIntervalMsProp, "60000") this.serverConfig.setProperty(KafkaConfig.NumReplicaFetchersProp, "1") @@ -58,6 +55,16 @@ class LogDirFailureTest extends IntegrationTestHarness { createTopic(topic, partitionNum, serverCount) } + override def createProducer: KafkaProducer[Array[Byte], Array[Byte]] = { + TestUtils.createProducer(brokerList, + retries = 0, + securityProtocol = this.securityProtocol, + trustStoreFile = this.trustStoreFile, + saslProperties = this.clientSaslProperties, + props = Some(producerConfig)) + } + + @Test def testIOExceptionDuringLogRoll() { testProduceAfterLogDirFailureOnLeader(Roll) @@ -175,7 +182,6 @@ class LogDirFailureTest extends IntegrationTestHarness { case t: NotLeaderForPartitionException => // This may happen if ProduceRequest version <= 3 case t: Throwable => fail(s"send() should fail with either KafkaStorageException or NotLeaderForPartitionException instead of ${t.toString}") } - case e: Throwable => fail(s"send() should fail with either KafkaStorageException or NotLeaderForPartitionException instead of ${e.toString}") } // Wait for producer to update metadata for the partition