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 dbb908d4cbf80..838d49fb67d2c 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 @@ -445,7 +445,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); @@ -468,7 +468,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, @@ -514,16 +515,9 @@ 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()) { + if (config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG)) { final String transactionalId = config.getString(ProducerConfig.TRANSACTIONAL_ID_CONFIG); final int transactionTimeoutMs = config.getInt(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG); final long retryBackoffMs = config.getLong(ProducerConfig.RETRY_BACKOFF_MS_CONFIG); @@ -543,28 +537,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..afc1e55cdfdad 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 @@ -27,6 +27,8 @@ import org.apache.kafka.common.config.SecurityConfig; import org.apache.kafka.common.metrics.Sensor; import org.apache.kafka.common.serialization.Serializer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Collections; import java.util.HashMap; @@ -44,6 +46,7 @@ * href="http://kafka.apache.org/documentation.html#producerconfigs">Kafka documentation */ public class ProducerConfig extends AbstractConfig { + private static final Logger log = LoggerFactory.getLogger(ProducerConfig.class); /* * NOTE: DO NOT CHANGE EITHER CONFIG STRINGS OR THEIR JAVA VARIABLE NAMES AS THESE ARE PART OF THE PUBLIC API AND @@ -104,7 +107,10 @@ public class ProducerConfig extends AbstractConfig { + "
  • acks=all This means the leader will wait for the full set of in-sync replicas to" + " acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica" + " remains alive. This is the strongest available guarantee. This is equivalent to the acks=-1 setting." - + ""; + + "" + + "

    " + + "Note that enabling idempotence requires this config value to be 'all'." + + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled."; /** linger.ms */ public static final String LINGER_MS_CONFIG = "linger.ms"; @@ -202,23 +208,32 @@ public class ProducerConfig extends AbstractConfig { /** metric.reporters */ public static final String METRIC_REPORTER_CLASSES_CONFIG = CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG; + // 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; + /** max.in.flight.requests.per.connection */ public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection"; 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)."; + + " message re-ordering after a failed send due to retries (i.e., if retries are enabled)." + + " Additionally, enabling idempotence requires this config value to be less than or equal to " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE + "." + + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled."; /** 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 additionally that produce requests will be" - + " failed before the number of retries has been exhausted if the timeout configured by" + + " 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."; + + " retry behavior." + + "

    " + + "Enabling idempotence requires this config value to be greater than 0." + + " If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled." + + "

    " + + "Allowing retries while setting enable.idempotence to false and " + 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."; /** key.serializer */ public static final String KEY_SERIALIZER_CLASS_CONFIG = "key.serializer"; @@ -269,10 +284,13 @@ 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 " - + 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."; + + "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'. " + + "

    " + + "Idempotence is enabled by default if no conflicting configurations are set. " + + "If conflicting configurations are set and idempotence is not explicitly enabled, idempotence is disabled. " + + "If idempotence is explicitly enabled and conflicting configurations are set, a ConfigException is thrown."; /** transaction.timeout.ms */ public static final String TRANSACTION_TIMEOUT_CONFIG = "transaction.timeout.ms"; @@ -438,9 +456,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 +473,56 @@ 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. - if (idempotenceEnabled()) { - boolean userConfiguredRetries = this.originals().containsKey(RETRIES_CONFIG); - if (this.getInt(RETRIES_CONFIG) == 0) { - throw new ConfigException("Must set " + ProducerConfig.RETRIES_CONFIG + " to non-zero when using the idempotent producer."); + final boolean userConfiguredIdempotence = this.originals().containsKey(ENABLE_IDEMPOTENCE_CONFIG); + boolean idempotenceEnabled = this.getBoolean(ENABLE_IDEMPOTENCE_CONFIG); + boolean shouldDisableIdempotence = false; + + // For idempotence producers, values for `retries` and `acks` and `max.in.flight.requests.per.connection` need validation + if (idempotenceEnabled) { + final int retries = this.getInt(RETRIES_CONFIG); + if (retries == 0) { + if (userConfiguredIdempotence) { + throw new ConfigException("Must set " + RETRIES_CONFIG + " to non-zero when using the idempotent producer."); + } + log.info("Idempotence will be disabled because {} is set to 0.", RETRIES_CONFIG, retries); + shouldDisableIdempotence = true; } - configs.put(RETRIES_CONFIG, userConfiguredRetries ? this.getInt(RETRIES_CONFIG) : Integer.MAX_VALUE); - boolean userConfiguredAcks = this.originals().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 " + + if (acks != (short) -1) { + if (userConfiguredIdempotence) { + throw new ConfigException("Must set " + ACKS_CONFIG + " to all in order to use the idempotent " + "producer. Otherwise we cannot guarantee idempotence."); + } + log.info("Idempotence will be disabled because {} is set to {}, not set to 'all'.", ACKS_CONFIG, acks); + shouldDisableIdempotence = true; + } + + final int inFlightConnection = this.getInt(MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION); + if (MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION_FOR_IDEMPOTENCE < inFlightConnection) { + if (userConfiguredIdempotence) { + throw new ConfigException("Must set " + MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION + " to at most 5" + + " to use the idempotent producer."); + } + log.warn("Idempotence will be disabled because {} is set to {}, which is greater than 5. " + + "Please note that in v4.0.0 and onward, this will become an error.", MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, inFlightConnection); + shouldDisableIdempotence = true; } - configs.put(ACKS_CONFIG, "-1"); + } + + if (shouldDisableIdempotence) { + configs.put(ENABLE_IDEMPOTENCE_CONFIG, false); + idempotenceEnabled = false; + } + + // validate `transaction.id` after validating idempotence dependant configs because `enable.idempotence` config might be overridden + boolean userConfiguredTransactions = originalConfigs.containsKey(TRANSACTIONAL_ID_CONFIG); + if (!idempotenceEnabled && userConfiguredTransactions) { + throw new ConfigException("Cannot set a " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " without also enabling idempotence."); } } @@ -513,16 +553,6 @@ public ProducerConfig(Map props) { super(CONFIG, 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) - throw new ConfigException("Cannot set a " + ProducerConfig.TRANSACTIONAL_ID_CONFIG + " without also enabling idempotence."); - return userConfiguredTransactions || idempotenceEnabled; - } - ProducerConfig(Map props, boolean doLog) { super(CONFIG, props, 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 1e45a58b98efe..f50e4a0c868a9 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 @@ -72,6 +72,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; @@ -122,8 +124,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, @@ -146,6 +147,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, @@ -226,6 +228,32 @@ public void testAcksAndIdempotenceForIdempotentProducers() { config.getString(ProducerConfig.ACKS_CONFIG), "acks should be overwritten"); + Properties validProps4 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.ACKS_CONFIG, "0"); + }}; + config = new ProducerConfig(validProps4); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be disabled when acks not set to all and `enable.idempotence` config is unset."); + assertEquals( + "0", + config.getString(ProducerConfig.ACKS_CONFIG), + "acks should be set with overridden value"); + + Properties validProps5 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.ACKS_CONFIG, "1"); + }}; + config = new ProducerConfig(validProps5); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be disabled when acks not set to all and `enable.idempotence` config is unset."); + assertEquals( + "1", + config.getString(ProducerConfig.ACKS_CONFIG), + "acks should be set with overridden value"); + Properties invalidProps = new Properties() {{ putAll(baseProps); setProperty(ProducerConfig.ACKS_CONFIG, "0"); @@ -240,6 +268,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( @@ -250,12 +279,149 @@ public void testAcksAndIdempotenceForIdempotentProducers() { Properties invalidProps3 = new Properties() {{ putAll(baseProps); setProperty(ProducerConfig.ACKS_CONFIG, "0"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps3), + "Must set acks to all when using the transactional 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 validProps2 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.RETRIES_CONFIG, "0"); + }}; + config = new ProducerConfig(validProps2); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be disabled when retries set to 0 and `enable.idempotence` config is unset."); + assertEquals( + 0, + config.getInt(ProducerConfig.RETRIES_CONFIG), + "retries should be set with overridden value"); + + 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"); + // explicitly enabling idempotence should still throw exception + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); + }}; + 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"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; + assertThrows( + ConfigException.class, + () -> new ProducerConfig(invalidProps3), + "Must set retries to non-zero when using the transactional 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, "6"); + setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false"); + }}; + ProducerConfig config = new ProducerConfig(validProps); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be overwritten"); + assertEquals( + 6, + config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION), + "max.in.flight.requests.per.connection should be overwritten"); + + Properties validProps2 = new Properties() {{ + putAll(baseProps); + setProperty(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "6"); + }}; + config = new ProducerConfig(validProps2); + assertFalse( + config.getBoolean(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG), + "idempotence should be disabled when `max.in.flight.requests.per.connection` is greater than 5 and " + + "`enable.idempotence` config is unset."); + assertEquals( + 6, + config.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION), + "`max.in.flight.requests.per.connection` should be set with overridden value"); + + 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, "6"); + // explicitly enabling idempotence should still throw exception setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); }}; + 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, "6"); + setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transactionalId"); + }}; assertThrows( ConfigException.class, () -> new ProducerConfig(invalidProps3), - "Must set acks to all in order to use the idempotent producer"); + "Must set retries to non-zero when using the idempotent producer."); } @Test @@ -467,9 +633,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 @@ -478,10 +652,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 @@ -511,18 +688,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); @@ -543,11 +716,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 with a partition higher than the initial (outdated) partition range ProducerRecord record = new ProducerRecord<>(topic, 2, null, "value"); @@ -568,6 +743,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()); @@ -581,11 +757,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"); @@ -605,11 +783,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"); @@ -630,7 +810,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(); @@ -648,6 +831,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; @@ -795,6 +980,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)); @@ -810,6 +998,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())); @@ -922,14 +1111,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)); @@ -955,7 +1144,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(), @@ -977,7 +1166,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)); @@ -999,7 +1188,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(), @@ -1036,10 +1225,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), @@ -1079,7 +1268,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(), @@ -1088,7 +1277,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)); @@ -1137,10 +1326,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; @@ -1193,7 +1382,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(), @@ -1404,7 +1593,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(); @@ -1433,7 +1622,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 7c221baf69a2b..4a6f7315eeeb1 100644 --- a/core/src/main/scala/kafka/tools/ConsoleProducer.scala +++ b/core/src/main/scala/kafka/tools/ConsoleProducer.scala @@ -164,7 +164,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 6efb860675d96..360d8f28d5be2 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) @@ -2376,6 +2377,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 687cd9e8b5685..9502d2ef24c85 100644 --- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala +++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala @@ -1670,6 +1670,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 73ebe2970a7da..6734d17c3796f 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -61,6 +61,14 @@

    Upgrading to 3.1.0 from any vers

    • +
    Notable changes in 3.1.1
    +
      +
    • Idempotence for the producer is enabled by default if no conflicting configurations are set. + A bug prevented the producer idempotence default from being applied which meant that it remained disabled unless the user had explicitly set + enable.idempotence to true. See KAFKA-13598for more details. + This issue was fixed and the default is properly applied.
      • +
    +
    Notable changes in 3.1.0
    • Apache Kafka supports Java 17.
      • @@ -124,8 +132,21 @@

      Upgrading to 3.0.0 from any vers +
      Notable changes in 3.0.1
      +
        +
      • Idempotence for the producer is enabled by default if no conflicting configurations are set. + A bug prevented the producer idempotence default from being applied which meant that it remained disabled unless the user had explicitly set + enable.idempotence to true. See KAFKA-13598for more details. + This issue was fixed and the default is properly applied.
        • +
      +

      Notable changes in 3.0.0
        +
      • The producer has stronger delivery guarantees by default: idempotence is enabled and acks is set to all instead of 1. + See KIP-679 for details. + In 3.0.0 and 3.1.0, a bug prevented the idempotence default from being applied which meant that it remained disabled unless the user had explicitly set + enable.idempotence to true. Note that the bug did not affect the acks=all change. See KAFKA-13598for more details. + This issue was fixed and the default is properly applied in 3.0.1, 3.1.1, and 3.2.0.
      • ZooKeeper has been upgraded to version 3.6.3.
      • A preview of KRaft mode is available, though upgrading to it from the 2.8 Early Access release is not possible. See the config/kraft/README.md file for details.
        • diff --git a/log4j-appender/src/test/java/org/apache/kafka/log4jappender/KafkaLog4jAppenderTest.java b/log4j-appender/src/test/java/org/apache/kafka/log4jappender/KafkaLog4jAppenderTest.java index 7ec56335fa3df..90a791f7f30cb 100644 --- a/log4j-appender/src/test/java/org/apache/kafka/log4jappender/KafkaLog4jAppenderTest.java +++ b/log4j-appender/src/test/java/org/apache/kafka/log4jappender/KafkaLog4jAppenderTest.java @@ -190,7 +190,7 @@ private Properties getLog4jConfigWithRealProducer(boolean ignoreExceptions) { props.put("log4j.appender.KAFKA.layout.ConversionPattern", "%-5p: %c - %m%n"); props.put("log4j.appender.KAFKA.BrokerList", "127.0.0.2:9093"); props.put("log4j.appender.KAFKA.Topic", "test-topic"); - props.put("log4j.appender.KAFKA.RequiredNumAcks", "1"); + props.put("log4j.appender.KAFKA.RequiredNumAcks", "-1"); props.put("log4j.appender.KAFKA.SyncSend", "true"); // setting producer timeout (max.block.ms) to be low props.put("log4j.appender.KAFKA.maxBlockMs", "10"); @@ -208,7 +208,7 @@ private Properties getLog4jConfig(boolean syncSend) { props.put("log4j.appender.KAFKA.layout.ConversionPattern", "%-5p: %c - %m%n"); props.put("log4j.appender.KAFKA.BrokerList", "127.0.0.1:9093"); props.put("log4j.appender.KAFKA.Topic", "test-topic"); - props.put("log4j.appender.KAFKA.RequiredNumAcks", "1"); + props.put("log4j.appender.KAFKA.RequiredNumAcks", "-1"); props.put("log4j.appender.KAFKA.SyncSend", Boolean.toString(syncSend)); props.put("log4j.logger.kafka.log4j", "INFO, KAFKA"); return props;