diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java index 58075d628927e..5371a73ece192 100644 --- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java +++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java @@ -194,8 +194,9 @@ public class CommonClientConfigs { public static Map postProcessReconnectBackoffConfigs(AbstractConfig config, Map parsedValues) { HashMap rval = new HashMap<>(); - if ((!config.originals().containsKey(RECONNECT_BACKOFF_MAX_MS_CONFIG)) && - config.originals().containsKey(RECONNECT_BACKOFF_MS_CONFIG)) { + Map originalConfig = config.originals(); + if ((!originalConfig.containsKey(RECONNECT_BACKOFF_MAX_MS_CONFIG)) && + originalConfig.containsKey(RECONNECT_BACKOFF_MS_CONFIG)) { log.debug("Disabling exponential reconnect backoff because {} is set, but {} is not.", RECONNECT_BACKOFF_MS_CONFIG, RECONNECT_BACKOFF_MAX_MS_CONFIG); rval.put(RECONNECT_BACKOFF_MAX_MS_CONFIG, parsedValues.get(RECONNECT_BACKOFF_MS_CONFIG)); 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 bedd6277e2c57..5fd058fc1bfaf 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 @@ -442,7 +442,7 @@ public KafkaProducer(Properties properties, Serializer keySerializer, Seriali // visible for testing Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) { - int maxInflightRequests = configureInflightRequests(producerConfig); + int maxInflightRequests = producerConfig.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG); ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(producerConfig, time, logContext); ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics); @@ -465,7 +465,8 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat apiVersions, throttleTimeSensor, logContext); - short acks = configureAcks(producerConfig, log); + + short acks = Short.parseShort(producerConfig.getString(ProducerConfig.ACKS_CONFIG)); return new Sender(logContext, client, metadata, @@ -511,15 +512,8 @@ private static int configureDeliveryTimeout(ProducerConfig config, Logger log) { private TransactionManager configureTransactionState(ProducerConfig config, LogContext logContext) { - TransactionManager transactionManager = null; - final boolean userConfiguredIdempotence = config.originals().containsKey(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG); - final boolean userConfiguredTransactions = config.originals().containsKey(ProducerConfig.TRANSACTIONAL_ID_CONFIG); - if (userConfiguredTransactions && !userConfiguredIdempotence) - log.info("Overriding the default {} to true since {} is specified.", ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, - ProducerConfig.TRANSACTIONAL_ID_CONFIG); - if (config.idempotenceEnabled()) { final String transactionalId = config.getString(ProducerConfig.TRANSACTIONAL_ID_CONFIG); final int transactionTimeoutMs = config.getInt(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG); @@ -540,28 +534,6 @@ private TransactionManager configureTransactionState(ProducerConfig config, return transactionManager; } - private static int configureInflightRequests(ProducerConfig config) { - if (config.idempotenceEnabled() && 5 < config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { - throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" + - " to use the idempotent producer."); - } - return config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); - } - - private static short configureAcks(ProducerConfig config, Logger log) { - boolean userConfiguredAcks = config.originals().containsKey(ProducerConfig.ACKS_CONFIG); - short acks = Short.parseShort(config.getString(ProducerConfig.ACKS_CONFIG)); - - if (config.idempotenceEnabled()) { - if (!userConfiguredAcks) - log.info("Overriding the default {} to all since idempotence is enabled.", ProducerConfig.ACKS_CONFIG); - else if (acks != -1) - throw new ConfigException("Must set " + ProducerConfig.ACKS_CONFIG + " to all in order to use the idempotent " + - "producer. Otherwise we cannot guarantee idempotence."); - } - return acks; - } - /** * Needs to be called before any other methods when the transactional.id is set in the configuration. * 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 fbd3449f29c8f..9a698b1ecef3c 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 @@ -207,6 +207,8 @@ public class ProducerConfig extends AbstractConfig { private static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_DOC = "The maximum number of unacknowledged requests the client will send on a single connection before blocking." + " Note that if this config is set to be greater than 1 and enable.idempotence is set to false, there is a risk of" + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)."; + // max.in.flight.requests.per.connection should be less than or equal to 5 when idempotence producer enabled to ensure message ordering + private static final int MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE = 5; /** retries */ public static final String RETRIES_CONFIG = CommonClientConfigs.RETRIES_CONFIG; @@ -269,8 +271,8 @@ public class ProducerConfig extends AbstractConfig { public static final String ENABLE_IDEMPOTENCE_CONFIG = "enable.idempotence"; public static final String ENABLE_IDEMPOTENCE_DOC = "When set to 'true', the producer will ensure that exactly one copy of each message is written in the stream. If 'false', producer " + "retries due to broker failures, etc., may write duplicates of the retried message in the stream. " - + "Note that enabling idempotence requires " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to be less than or equal to 5 " - + "(with message ordering preserved for any allowable value), " + RETRIES_CONFIG + " to be greater than 0, and " + + "Note that enabling idempotence requires " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + + " (with message ordering preserved for any allowable value), " + RETRIES_CONFIG + " to be greater than 0, and " + ACKS_CONFIG + " must be 'all'. If these values are not explicitly set by the user, suitable values will be chosen. If incompatible " + "values are set, a ConfigException will be thrown."; @@ -438,9 +440,8 @@ public class ProducerConfig extends AbstractConfig { @Override protected Map postProcessParsedConfig(final Map parsedValues) { Map refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues); - maybeOverrideEnableIdempotence(refinedConfigs); + postProcessAndValidateIdempotenceConfigs(refinedConfigs); maybeOverrideClientId(refinedConfigs); - maybeOverrideAcksAndRetries(refinedConfigs); return refinedConfigs; } @@ -456,33 +457,30 @@ private void maybeOverrideClientId(final Map configs) { configs.put(CLIENT_ID_CONFIG, refinedClientId); } - private void maybeOverrideEnableIdempotence(final Map configs) { - boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG); - boolean userConfiguredTransactions = this.originals().containsKey(TRANSACTIONAL_ID_CONFIG); - - if (userConfiguredTransactions && !userConfiguredIdempotence) { - configs.put(ENABLE_IDEMPOTENCE_CONFIG, true); - } - } - - private void maybeOverrideAcksAndRetries(final Map configs) { + private void postProcessAndValidateIdempotenceConfigs(final Map configs) { + final Map originalConfigs = this.originals(); final String acksStr = parseAcks(this.getString(ACKS_CONFIG)); configs.put(ACKS_CONFIG, acksStr); - // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` might need to be overridden. + + // For idempotence producers, values for `RETRIES_CONFIG` and `ACKS_CONFIG` need validation if (idempotenceEnabled()) { - boolean userConfiguredRetries = this.originals().containsKey(RETRIES_CONFIG); - if (this.getInt(RETRIES_CONFIG) == 0) { + boolean userConfiguredRetries = originalConfigs.containsKey(RETRIES_CONFIG); + if (userConfiguredRetries && this.getInt(RETRIES_CONFIG) == 0) { throw new ConfigException("Must set " + ProducerConfig.RETRIES_CONFIG + " to non-zero when using the idempotent producer."); } - configs.put(RETRIES_CONFIG, userConfiguredRetries ? this.getInt(RETRIES_CONFIG) : Integer.MAX_VALUE); - boolean userConfiguredAcks = this.originals().containsKey(ACKS_CONFIG); + boolean userConfiguredAcks = originalConfigs.containsKey(ACKS_CONFIG); final short acks = Short.valueOf(acksStr); if (userConfiguredAcks && acks != (short) -1) { throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " + "producer. Otherwise we cannot guarantee idempotence."); } - configs.put(ACKS_CONFIG, "-1"); + + boolean userConfiguredInflightRequests = originalConfigs.containsKey(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); + if (userConfiguredInflightRequests && MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < this.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION)) { + throw new ConfigException("Must set " + ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" + + " to use the idempotent producer."); + } } } @@ -514,13 +512,12 @@ public ProducerConfig(Map props) { } boolean idempotenceEnabled() { - boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG); boolean userConfiguredTransactions = this.originals().containsKey(TRANSACTIONAL_ID_CONFIG); - boolean idempotenceEnabled = userConfiguredIdempotence && this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG); - - if (!idempotenceEnabled && userConfiguredIdempotence && userConfiguredTransactions) + boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG); + if (!idempotenceEnabled && userConfiguredTransactions) throw new ConfigException("Cannot set a " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " without also enabling idempotence."); - return userConfiguredTransactions || idempotenceEnabled; + + return idempotenceEnabled; } ProducerConfig(Map props, boolean doLog) { diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java index 76e920b60877f..7221d37e5cdf3 100644 --- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java @@ -73,6 +73,8 @@ import org.apache.kafka.test.MockSerializer; import org.apache.kafka.test.TestUtils; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import javax.management.MBeanServer; import javax.management.ObjectName; @@ -125,8 +127,7 @@ public class KafkaProducerTest { private final String topic = "topic"; - private final Node host1 = new Node(0, "host1", 1000); - private final Collection nodes = Collections.singletonList(host1); + private final Collection nodes = Collections.singletonList(NODE); private final Cluster emptyCluster = new Cluster( null, nodes, @@ -149,6 +150,7 @@ public class KafkaProducerTest { Collections.emptySet(), Collections.emptySet()); private static final int DEFAULT_METADATA_IDLE_MS = 5 * 60 * 1000; + private static final Node NODE = new Node(0, "host1", 1000); private static KafkaProducer kafkaProducer(Map configs, @@ -243,6 +245,7 @@ public void testAcksAndIdempotenceForIdempotentProducers() { Properties invalidProps2 = new Properties() {{ putAll(baseProps); setProperty(ProducerConfig.ACKS_CONFIG, "1"); + // explicitly enabling idempotence should still throw exception setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); }}; assertThrows( @@ -253,12 +256,149 @@ public void testAcksAndIdempotenceForIdempotentProducers() { Properties invalidProps3 = new Properties() {{ putAll(baseProps); setProperty(ProducerConfig.ACKS_CONFIG, "0"); - setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); }}; assertThrows( ConfigException.class, () -> new ProducerConfig(invalidProps3), "Must set acks to all in order to use the idempotent producer"); + + Properties invalidProps4 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.ACKS_CONFIG, "0"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps4), + "Must set retries to non-zero when using the idempotent producer."); + } + + @Test + public void testRetriesAndIdempotenceForIdempotentProducers() { + Properties baseProps = new Properties() {{ + setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + }}; + + Properties validProps = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + }}; + ProducerConfig config = new ProducerConfig(validProps); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be overwritten"); + assertEquals( + 0, + config.getInt(ProducerConfig.RETRIES_CONFIG), + "retries should be overwritten"); + + Properties invalidProps = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps), + "Cannot set a transactional.id without also enabling idempotence"); + + Properties invalidProps2 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps2), + "Must set retries to non-zero when using the idempotent producer."); + + Properties invalidProps3 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + // explicitly enabling idempotence should still throw exception + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps3), + "Must set retries to non-zero when using the idempotent producer."); + + Properties invalidProps4 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps4), + "Must set retries to non-zero when using the idempotent producer."); + } + + @Test + public void testInflightRequestsAndIdempotenceForIdempotentProducers() { + Properties baseProps = new Properties() {{ + setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + }}; + + Properties validProps = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10"); + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + }}; + ProducerConfig config = new ProducerConfig(validProps); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be overwritten"); + assertEquals( + 10, + config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION), + "max.in.flight.requests.per.connection should be overwritten"); + + Properties invalidProps = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5"); + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps), + "Cannot set a transactional.id without also enabling idempotence"); + + Properties invalidProps2 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps2), + "Must set max.in.flight.requests.per.connection to at most 5 when using the idempotent producer."); + + Properties invalidProps3 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10"); + // explicitly enabling idempotence should still throw exception + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps3), + "Must set max.in.flight.requests.per.connection to at most 5 when using the idempotent producer."); + + Properties invalidProps4 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "10"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps4), + "Must set retries to non-zero when using the idempotent producer."); } @Test @@ -470,9 +610,17 @@ private static KafkaProducer producerWithOverrideNewSender(Map producerWithOverrideNewSender(Map configs, ProducerMetadata metadata, Time timer) { + // let mockClient#leastLoadedNode return the node directly so that we can isolate Metadata calls from KafkaProducer for idempotent producer + MockClient mockClient = new MockClient(Time.SYSTEM, metadata) { + @Override + public Node leastLoadedNode(long now) { + return NODE; + } + }; + return new KafkaProducer( new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, new StringSerializer(), new StringSerializer())), - new StringSerializer(), new StringSerializer(), metadata, new MockClient(Time.SYSTEM, metadata), null, timer) { + new StringSerializer(), new StringSerializer(), metadata, mockClient, null, timer) { @Override Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) { // give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer @@ -481,10 +629,13 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat }; } - @Test - public void testMetadataFetch() throws InterruptedException { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMetadataFetch(boolean isIdempotenceEnabled) throws InterruptedException { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled); + ProducerMetadata metadata = mock(ProducerMetadata.class); // Return empty cluster 4 times and cluster from then on @@ -514,18 +665,14 @@ public void testMetadataFetch() throws InterruptedException { producer.close(Duration.ofMillis(0)); } - @Test - public void testMetadataExpiry() throws InterruptedException { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMetadataExpiry(boolean isIdempotenceEnabled) throws InterruptedException { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled); ProducerMetadata metadata = mock(ProducerMetadata.class); - Cluster emptyCluster = new Cluster( - "dummy", - Collections.singletonList(host1), - Collections.emptySet(), - Collections.emptySet(), - Collections.emptySet()); when(metadata.fetch()).thenReturn(onePartitionCluster, emptyCluster, onePartitionCluster); KafkaProducer producer = producerWithOverrideNewSender(configs, metadata); @@ -546,11 +693,13 @@ public void testMetadataExpiry() throws InterruptedException { producer.close(Duration.ofMillis(0)); } - @Test - public void testMetadataTimeoutWithMissingTopic() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMetadataTimeoutWithMissingTopic(boolean isIdempotenceEnabled) throws Exception { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000); + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled); // Create a record for a not-yet-created topic ProducerRecord record = new ProducerRecord<>(topic, 2, null, "value"); @@ -571,6 +720,7 @@ public void testMetadataTimeoutWithMissingTopic() throws Exception { // Four request updates where the topic isn't present, at which point the timeout expires and a // TimeoutException is thrown + // For idempotence enabled case, the first metadata.fetch will be called in Sender#maybeSendAndPollTransactionalRequest Future future = producer.send(record); verify(metadata, times(4)).requestUpdateForTopic(topic); verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong()); @@ -584,11 +734,13 @@ public void testMetadataTimeoutWithMissingTopic() throws Exception { } } - @Test - public void testMetadataWithPartitionOutOfRange() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMetadataWithPartitionOutOfRange(boolean isIdempotenceEnabled) throws Exception { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000); + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled); // Create a record with a partition higher than the initial (outdated) partition range ProducerRecord record = new ProducerRecord<>(topic, 2, null, "value"); @@ -608,11 +760,13 @@ public void testMetadataWithPartitionOutOfRange() throws Exception { producer.close(Duration.ofMillis(0)); } - @Test - public void testMetadataTimeoutWithPartitionOutOfRange() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testMetadataTimeoutWithPartitionOutOfRange(boolean isIdempotenceEnabled) throws Exception { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 60000); + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled); // Create a record with a partition higher than the initial (outdated) partition range ProducerRecord record = new ProducerRecord<>(topic, 2, null, "value"); @@ -633,7 +787,10 @@ public void testMetadataTimeoutWithPartitionOutOfRange() throws Exception { // Four request updates where the requested partition is out of range, at which point the timeout expires // and a TimeoutException is thrown + // For idempotence enabled case, the first and last metadata.fetch will be called in Sender#maybeSendAndPollTransactionalRequest, + // before the producer#send and after it finished Future future = producer.send(record); + verify(metadata, times(4)).requestUpdateForTopic(topic); verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong()); verify(metadata, times(5)).fetch(); @@ -651,6 +808,8 @@ public void testTopicRefreshInMetadata() throws InterruptedException { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "600000"); + // test under normal producer for simplicity + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false); long refreshBackoffMs = 500L; long metadataExpireMs = 60000L; long metadataIdleMs = 60000L; @@ -798,6 +957,9 @@ public void closeWithNegativeTimestampShouldThrow() { public void testFlushCompleteSendOfInflightBatches() { Map configs = new HashMap<>(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9000"); + // only test in idempotence disabled producer for simplicity + // flush operation acts the same for idempotence enabled and disabled cases + configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false); Time time = new MockTime(1); MetadataResponse initialUpdateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap("topic", 1)); @@ -813,6 +975,7 @@ public void testFlushCompleteSendOfInflightBatches() { Future response = producer.send(new ProducerRecord<>("topic", "value" + i)); futureResponses.add(response); } + futureResponses.forEach(res -> assertFalse(res.isDone())); producer.flush(); futureResponses.forEach(res -> assertTrue(res.isDone())); @@ -930,7 +1093,7 @@ public void testInitTransactionsResponseAfterTimeout() throws Exception { client.prepareResponse( request -> request instanceof FindCoordinatorRequest && ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(), - FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", host1)); + FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE)); Future future = executor.submit(producer::initTransactions); TestUtils.waitForCondition(client::hasInFlightRequests, @@ -967,14 +1130,14 @@ public void testInitTransactionTimeout() { client.prepareResponse( request -> request instanceof FindCoordinatorRequest && ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(), - FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", host1)); + FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE)); assertThrows(TimeoutException.class, producer::initTransactions); client.prepareResponse( request -> request instanceof FindCoordinatorRequest && ((FindCoordinatorRequest) request).data().keyType() == FindCoordinatorRequest.CoordinatorType.TRANSACTION.id(), - FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", host1)); + FindCoordinatorResponse.prepareResponse(Errors.NONE, "bad-transaction", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); @@ -1000,7 +1163,7 @@ public void testInitTransactionWhileThrottled() { Node node = metadata.fetch().nodes().get(0); client.throttle(node, 5000); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); try (Producer producer = kafkaProducer(configs, new StringSerializer(), @@ -1022,7 +1185,7 @@ public void testAbortTransaction() { MockClient client = new MockClient(time, metadata); client.updateMetadata(initialUpdateResponse); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); client.prepareResponse(endTxnResponse(Errors.NONE)); @@ -1044,7 +1207,7 @@ public void testMeasureAbortTransactionDuration() { ProducerMetadata metadata = newMetadata(0, Long.MAX_VALUE); MockClient client = new MockClient(time, metadata); client.updateMetadata(initialUpdateResponse); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); try (KafkaProducer producer = kafkaProducer(configs, new StringSerializer(), @@ -1075,7 +1238,7 @@ public void testCommitTransactionWithRecordTooLargeException() throws Exception ProducerMetadata metadata = mock(ProducerMetadata.class); MockClient client = new MockClient(time, metadata); client.updateMetadata(initialUpdateResponse); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); when(metadata.fetch()).thenReturn(onePartitionCluster); @@ -1108,7 +1271,7 @@ public void testCommitTransactionWithMetadataTimeoutForMissingTopic() throws Exc MockTime mockTime = new MockTime(); MockClient client = new MockClient(mockTime, metadata); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); AtomicInteger invocationCount = new AtomicInteger(0); @@ -1145,7 +1308,7 @@ public void testCommitTransactionWithMetadataTimeoutForPartitionOutOfRange() thr MockTime mockTime = new MockTime(); MockClient client = new MockClient(mockTime, metadata); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); AtomicInteger invocationCount = new AtomicInteger(0); @@ -1181,7 +1344,7 @@ public void testCommitTransactionWithSendToInvalidTopic() throws Exception { metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false, time.milliseconds()); MockClient client = new MockClient(time, metadata); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); String invalidTopicName = "topic abc"; // Invalid topic name due to space @@ -1224,10 +1387,10 @@ public void testSendTxnOffsetsWithGroupId() { Node node = metadata.fetch().nodes().get(0); client.throttle(node, 5000); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE)); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); String groupId = "group"; client.prepareResponse(request -> ((TxnOffsetCommitRequest) request).data().groupId().equals(groupId), @@ -1267,7 +1430,7 @@ public void testMeasureTransactionDurations() { MockClient client = new MockClient(time, metadata); client.updateMetadata(initialUpdateResponse); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); try (KafkaProducer producer = kafkaProducer(configs, new StringSerializer(), @@ -1276,7 +1439,7 @@ public void testMeasureTransactionDurations() { assertDurationAtLeast(producer, "txn-init-time-ns-total", tick.toNanos()); client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE)); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(txnOffsetsCommitResponse(Collections.singletonMap( new TopicPartition("topic", 0), Errors.NONE))); client.prepareResponse(endTxnResponse(Errors.NONE)); @@ -1325,10 +1488,10 @@ public void testSendTxnOffsetsWithGroupMetadata() { Node node = metadata.fetch().nodes().get(0); client.throttle(node, 5000); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); client.prepareResponse(addOffsetsToTxnResponse(Errors.NONE)); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); String groupId = "group"; String memberId = "member"; int generationId = 5; @@ -1381,7 +1544,7 @@ private void verifyInvalidGroupMetadata(ConsumerGroupMetadata groupMetadata) { Node node = metadata.fetch().nodes().get(0); client.throttle(node, 5000); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "some.id", NODE)); client.prepareResponse(initProducerIdResponse(1L, (short) 5, Errors.NONE)); try (Producer producer = kafkaProducer(configs, new StringSerializer(), @@ -1592,7 +1755,7 @@ public void testCloseIsForcedOnPendingInitProducerId() throws InterruptedExcepti ExecutorService executorService = Executors.newSingleThreadExecutor(); CountDownLatch assertionDoneLatch = new CountDownLatch(1); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", NODE)); executorService.submit(() -> { assertThrows(KafkaException.class, producer::initTransactions); assertionDoneLatch.countDown(); @@ -1621,7 +1784,7 @@ public void testCloseIsForcedOnPendingAddOffsetRequest() throws InterruptedExcep ExecutorService executorService = Executors.newSingleThreadExecutor(); CountDownLatch assertionDoneLatch = new CountDownLatch(1); - client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", host1)); + client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE, "this-is-a-transactional-id", NODE)); executorService.submit(() -> { assertThrows(KafkaException.class, producer::initTransactions); assertionDoneLatch.countDown(); 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 44902c0df1abc..eb28102e1001c 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 @@ -170,6 +170,7 @@ public void configure(final WorkerConfig config) { 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.ENABLE_IDEMPOTENCE_CONFIG, false); // disable idempotence since retries is force to 0 ConnectUtils.addMetricsContextProperties(producerProps, config, clusterId); Map consumerProps = new HashMap<>(originals); diff --git a/core/src/main/scala/kafka/tools/ConsoleProducer.scala b/core/src/main/scala/kafka/tools/ConsoleProducer.scala index c3a26fbe3965c..21ee0f4b8c048 100644 --- a/core/src/main/scala/kafka/tools/ConsoleProducer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleProducer.scala @@ -168,7 +168,7 @@ object ConsoleProducer { .withRequiredArg .describedAs("request required acks") .ofType(classOf[java.lang.String]) - .defaultsTo("1") + .defaultsTo("-1") val requestTimeoutMsOpt = parser.accepts("request-timeout-ms", "The ack timeout of the producer requests. Value must be non-negative and non-zero.") .withRequiredArg .describedAs("request timeout ms") diff --git a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala index 1da4fcd587edd..0b215ff7200bb 100644 --- a/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/AuthorizerIntegrationTest.scala @@ -143,6 +143,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest { val adminClients = Buffer[Admin]() producerConfig.setProperty(ProducerConfig.ACKS_CONFIG, "1") + producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") producerConfig.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "50000") consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, group) @@ -2383,6 +2384,7 @@ class AuthorizerIntegrationTest extends BaseRequestTest { private def buildTransactionalProducer(): KafkaProducer[Array[Byte], Array[Byte]] = { producerConfig.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId) + producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true") producerConfig.setProperty(ProducerConfig.ACKS_CONFIG, "all") createProducer() } diff --git a/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala index e8c451ec3aba2..0ce4004c64e2b 100644 --- a/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/ConsumerWithLegacyMessageFormatIntegrationTest.scala @@ -18,6 +18,7 @@ package kafka.api import kafka.log.LogConfig import kafka.server.KafkaConfig +import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.common.TopicPartition import org.junit.jupiter.api.Assertions.{assertEquals, assertNull, assertThrows} import org.junit.jupiter.api.Test @@ -101,7 +102,10 @@ class ConsumerWithLegacyMessageFormatIntegrationTest extends AbstractConsumerTes def testEarliestOrLatestOffsets(): Unit = { val topic0 = "topicWithNewMessageFormat" val topic1 = "topicWithOldMessageFormat" - val producer = createProducer() + val prop = new Properties() + // idempotence producer doesn't support old version of messages + prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") + val producer = createProducer(configOverrides = prop) createTopicAndSendRecords(producer, topicName = topic0, numPartitions = 2, recordsPerPartition = 100) val props = new Properties() props.setProperty(LogConfig.MessageFormatVersionProp, "0.9.0") diff --git a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala index cbb536b37b3ad..fb8431253f087 100644 --- a/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala +++ b/core/src/test/scala/integration/kafka/api/EndToEndAuthorizationTest.scala @@ -20,7 +20,7 @@ package kafka.api import com.yammer.metrics.core.Gauge import java.io.File -import java.util.Collections +import java.util.{Collections, Properties} import java.util.concurrent.ExecutionException import kafka.admin.AclCommand import kafka.metrics.KafkaYammerMetrics @@ -30,7 +30,7 @@ import kafka.server._ import kafka.utils._ import org.apache.kafka.clients.admin.Admin import org.apache.kafka.clients.consumer.{Consumer, ConsumerConfig, ConsumerRecords} -import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord} +import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord} import org.apache.kafka.common.acl._ import org.apache.kafka.common.acl.AclOperation._ import org.apache.kafka.common.acl.AclPermissionType._ @@ -42,6 +42,8 @@ import org.apache.kafka.common.resource.PatternType.{LITERAL, PREFIXED} import org.apache.kafka.common.security.auth.KafkaPrincipal import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.{AfterEach, BeforeEach, Test, TestInfo} +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource import scala.jdk.CollectionConverters._ @@ -334,13 +336,18 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas * messages and describe topics respectively when the describe ACL isn't set. * Also verifies that subsequent publish, consume and describe to authorized topic succeeds. */ - @Test - def testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl(): Unit = { + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl(isIdempotenceEnabled: Boolean): Unit = { // Set consumer group acls since we are testing topic authorization setConsumerGroupAcls() // Verify produce/consume/describe throw TopicAuthorizationException - val producer = createProducer() + + val prop = new Properties() + prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled.toString) + val producer = createProducer(configOverrides = prop) + assertThrows(classOf[TopicAuthorizationException], () => sendRecords(producer, numRecords, tp)) val consumer = createConsumer() consumer.assign(List(tp).asJava) @@ -352,12 +359,21 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas // Verify successful produce/consume/describe on another topic using the same producer, consumer and adminClient val topic2 = "topic2" val tp2 = new TopicPartition(topic2, 0) + setReadAndWriteAcls(tp2) - sendRecords(producer, numRecords, tp2) + // in idempotence producer, we need to create another producer because the previous one is in FATAL_ERROR state (due to authorization error) + // If the transaction state in FATAL_ERROR, it'll never transit to other state. check TransactionManager#isTransitionValid for detail + val producer2 = if (isIdempotenceEnabled) + createProducer(configOverrides = prop) + else + producer + + sendRecords(producer2, numRecords, tp2) consumer.assign(List(tp2).asJava) consumeRecords(consumer, numRecords, topic = topic2) val describeResults = adminClient.describeTopics(Set(topic, topic2).asJava).topicNameValues() assertEquals(1, describeResults.get(topic2).get().partitions().size()) + val e2 = assertThrows(classOf[ExecutionException], () => adminClient.describeTopics(Set(topic).asJava).allTopicNames().get()) assertTrue(e2.getCause.isInstanceOf[TopicAuthorizationException], "Unexpected exception " + e2.getCause) @@ -365,7 +381,7 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas // from the unauthorized topic and throw; since we can now return data during the time we are updating // metadata / fetching positions, it is possible that the authorized topic record is returned during this time. consumer.assign(List(tp, tp2).asJava) - sendRecords(producer, numRecords, tp2) + sendRecords(producer2, numRecords, tp2) var topic2RecordConsumed = false def verifyNoRecords(records: ConsumerRecords[Array[Byte], Array[Byte]]): Boolean = { assertEquals(Collections.singleton(tp2), records.partitions(), "Consumed records with unexpected partitions: " + records) @@ -380,22 +396,32 @@ abstract class EndToEndAuthorizationTest extends IntegrationTestHarness with Sas if (!topic2RecordConsumed) { consumeRecordsIgnoreOneAuthorizationException(consumer, numRecords, startingOffset = 1, topic2) } - sendRecords(producer, numRecords, tp) + sendRecords(producer2, numRecords, tp) consumeRecordsIgnoreOneAuthorizationException(consumer, numRecords, startingOffset = 0, topic) val describeResults2 = adminClient.describeTopics(Set(topic, topic2).asJava).topicNameValues assertEquals(1, describeResults2.get(topic).get().partitions().size()) assertEquals(1, describeResults2.get(topic2).get().partitions().size()) } - @Test - def testNoProduceWithDescribeAcl(): Unit = { + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testNoProduceWithDescribeAcl(isIdempotenceEnabled: Boolean): Unit = { AclCommand.main(describeAclArgs) servers.foreach { s => TestUtils.waitAndVerifyAcls(TopicDescribeAcl, s.dataPlaneRequestProcessor.authorizer.get, topicResource) } - val producer = createProducer() - val e = assertThrows(classOf[TopicAuthorizationException], () => sendRecords(producer, numRecords, tp)) - assertEquals(Set(topic).asJava, e.unauthorizedTopics()) + + val prop = new Properties() + prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled.toString) + val producer = createProducer(configOverrides = prop) + + if (isIdempotenceEnabled) { + // in idempotent producer, it'll fail at InitProducerId request + assertThrows(classOf[KafkaException], () => sendRecords(producer, numRecords, tp)) + } else { + val e = assertThrows(classOf[TopicAuthorizationException], () => sendRecords(producer, numRecords, tp)) + assertEquals(Set(topic).asJava, e.unauthorizedTopics()) + } confirmReauthenticationMetrics() } diff --git a/core/src/test/scala/integration/kafka/api/MetricsTest.scala b/core/src/test/scala/integration/kafka/api/MetricsTest.scala index 850ac89365af2..151bd0dac3358 100644 --- a/core/src/test/scala/integration/kafka/api/MetricsTest.scala +++ b/core/src/test/scala/integration/kafka/api/MetricsTest.scala @@ -82,7 +82,10 @@ class MetricsTest extends IntegrationTestHarness with SaslSetup { // Produce and consume some records val numRecords = 10 val recordSize = 100000 - val producer = createProducer() + val prop = new Properties() + // idempotence producer doesn't support old version of messages + prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") + val producer = createProducer(configOverrides = prop) sendRecords(producer, numRecords, recordSize, tp) val consumer = createConsumer() diff --git a/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala index a9f2c6c4535b8..c9be59e6b41fc 100644 --- a/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala +++ b/core/src/test/scala/integration/kafka/api/SaslClientsWithInvalidCredentialsTest.scala @@ -14,9 +14,8 @@ package kafka.api import java.nio.file.Files import java.time.Duration -import java.util.Collections +import java.util.{Collections, Properties} import java.util.concurrent.{ExecutionException, TimeUnit} - import scala.jdk.CollectionConverters._ import org.apache.kafka.clients.admin.{Admin, AdminClientConfig} import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer} @@ -30,6 +29,8 @@ import kafka.server.KafkaConfig import kafka.utils.{JaasTestUtils, TestUtils} import kafka.zk.ConfigEntityChangeNotificationZNode import org.apache.kafka.common.security.auth.SecurityProtocol +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.ValueSource class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with SaslSetup { private val kafkaClientSaslMechanism = "SCRAM-SHA-256" @@ -76,14 +77,24 @@ class SaslClientsWithInvalidCredentialsTest extends IntegrationTestHarness with closeSasl() } - @Test - def testProducerWithAuthenticationFailure(): Unit = { - val producer = createProducer() + @ParameterizedTest + @ValueSource(booleans = Array(true, false)) + def testProducerWithAuthenticationFailure(isIdempotenceEnabled: Boolean): Unit = { + val prop = new Properties() + prop.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, isIdempotenceEnabled.toString) + val producer = createProducer(configOverrides = prop) + verifyAuthenticationException(sendOneRecord(producer, maxWaitMs = 10000)) verifyAuthenticationException(producer.partitionsFor(topic)) createClientCredential() - verifyWithRetry(sendOneRecord(producer)) + // in idempotence producer, we need to create another producer because the previous one is in FATEL_ERROR state (due to authentication error) + // If the transaction state in FATAL_ERROR, it'll never transit to other state. check TransactionManager#isTransitionValid for detail + val producer2 = if (isIdempotenceEnabled) + createProducer(configOverrides = prop) + else + producer + verifyWithRetry(sendOneRecord(producer2)) } @Test diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala index 1dd4096b8bf9a..0cc58628103da 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -1678,6 +1678,8 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup 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) + // disable the idempotence since some tests want to test the cases when retries=0, and these tests are not testing producers + producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") val producer = new KafkaProducer[String, String](producerProps, new StringSerializer, new StringSerializer) producers += producer diff --git a/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala b/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala index 1025d7ae2e3bd..bfbb14e1aaae7 100644 --- a/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala +++ b/core/src/test/scala/unit/kafka/server/LogDirFailureTest.scala @@ -109,6 +109,7 @@ class LogDirFailureTest extends IntegrationTestHarness { @Test def testReplicaFetcherThreadAfterLogDirFailureOnFollower(): Unit = { this.producerConfig.setProperty(ProducerConfig.RETRIES_CONFIG, "0") + this.producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") val producer = createProducer() val partition = new TopicPartition(topic, 0) @@ -140,6 +141,7 @@ class LogDirFailureTest extends IntegrationTestHarness { def testProduceErrorsFromLogDirFailureOnLeader(failureType: LogDirFailureType): Unit = { // Disable retries to allow exception to bubble up for validation this.producerConfig.setProperty(ProducerConfig.RETRIES_CONFIG, "0") + this.producerConfig.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false") val producer = createProducer() val partition = new TopicPartition(topic, 0) diff --git a/docs/upgrade.html b/docs/upgrade.html index 459ac0a3b4058..e0c9be3b0fd26 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -19,6 +19,14 @@