GH-1957: Align EOSMode with KIP-732

Resolves #1957

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/appendix.adoc

@@ -2,10 +2,12 @@
 == Override Spring Boot Dependencies
 
 When using Spring for Apache Kafka in a Spring Boot application, the Apache Kafka dependency versions are determined by Spring Boot's dependency management.
-If you wish to use a different version of `kafka-clients` or `kafka-streams`, such as 2.8.0, you need to override all of the associated dependencies.
+If you wish to use a different version of `kafka-clients` or `kafka-streams`, such as 3.0.0, you need to override all of the associated dependencies.
 This is especially true when using the embedded Kafka broker in `spring-kafka-test`.
 
-IMPORTANT: Backwards compatibility is not supported for all Boot versions; Spring for Apache Kafka 2.7 has been tested with Spring Boot 2.4 and 2.5.
+IMPORTANT: Backwards compatibility is not supported for all Boot versions; Spring for Apache Kafka 2.8 has been tested with Spring Boot 2.5 and 2.6.
+
+IMPORTANT: When using the 3.0.0 clients, the default `EOSMode.V2` is not supported unless the broker is 2.5 or higher; if you wish to use the 3.0.0 clients with an earlier broker, you must change the container's `EOSMode` to `V1`.
 
 ====
 [source, xml, subs="+attributes", role="primary"]

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2462,7 +2462,7 @@ The default executor creates threads named `<name>-C-n`; with the `KafkaMessageL
 |See <<delivery-header>>.
 
 |[[eosMode]]<<eosMode,`eosMode`>>
-|`BETA`
+|`V2`
 |Exactly Once Semantics mode; see <<exactly-once>>.
 
 |[[fixTxOffsets]]<<fixTxOffsets,`fixTxOffsets`>>
@@ -3534,33 +3534,40 @@ This means that, for a `read->process-write` sequence, it is guaranteed that the
 
 Spring for Apache Kafka version 2.5 and later supports two EOS modes:
 
-* `ALPHA` - aka `transactional.id` fencing (since version 0.11.0.0)
-* `BETA` - aka fetch-offset-request fencing (since version 2.5)
+* `ALPHA` - alias for `V1` (deprecated)
+* `BETA` - alias for `V2` (deprecated)
+* `V1` - aka `transactional.id` fencing (since version 0.11.0.0)
+* `V2` - aka fetch-offset-request fencing (since version 2.5)
 
-With mode `ALPHA`, the producer is "fenced" if another instance with the same `transactional.id` is started.
+With mode `V1`, the producer is "fenced" if another instance with the same `transactional.id` is started.
 Spring manages this by using a `Producer` for each `group.id/topic/partition`; when a rebalance occurs a new instance will use the same `transactional.id` and the old producer is fenced.
 
-With mode `BETA`, it is not necessary to have a producer for each `group.id/topic/partition` because consumer metadata is sent along with the offsets to the transaction and the broker can determine if the producer is fenced using that information instead.
+With mode `V2`, it is not necessary to have a producer for each `group.id/topic/partition` because consumer metadata is sent along with the offsets to the transaction and the broker can determine if the producer is fenced using that information instead.
 
-Starting with version 2.6, the default `EOSMode` is `BETA`.
+Starting with version 2.6, the default `EOSMode` is `V2`.
 
 To configure the container to use mode `ALPHA`, set the container property `EOSMode` to `ALPHA`, to revert to the previous behavior.
 
-IMPORTANT: With `BETA`, your brokers must be version 2.5 or later, however with `kafka-clients` version 2.6, the producer will automatically fall back to `ALPHA` if the broker does not support `BETA`.
+IMPORTANT: With `V2`, your brokers must be version 2.5 or later, however with `kafka-clients` version 2.6, the producer will automatically fall back to `ALPHA` if the broker does not support `BETA`.
 The `DefaultKafkaProducerFactory` is configured to enable that behavior.
 If your brokers are earlier than 2.5, be sure to leave the `DefaultKafkaProducerFactory` `producerPerConsumerPartition` set to `true` and, if you are using a batch listener, you should set `subBatchPerPartition` to `true`.
 
-When your brokers are upgraded to 2.5 or later, the producer will automatically switch to using mode `BETA`, but the number of producers will remain as before.
+When your brokers are upgraded to 2.5 or later, the producer will automatically use mode `V2`, but the number of producers will remain as before.
 You can then do a rolling upgrade of your application with `producerPerConsumerPartition` set to `false` to reduce the number of producers; you should also no longer set the `subBatchPerPartition` container property.
 
 If your brokers are already 2.5 or newer, you should set the `DefaultKafkaProducerFactory` `producerPerConsumerPartition` property to `false`, to reduce the number of producers needed.
 
-IMPORTANT: When using `EOSMode.BETA` with `producerPerConsumerPartition=false` the `transactional.id` must be unique across all application instances.
+IMPORTANT: When using `EOSMode.V2` with `producerPerConsumerPartition=false` the `transactional.id` must be unique across all application instances.
 
-When using `BETA` mode, it is no longer necessary to set the `subBatchPerPartition` to `true`; it will default to `false` when the `EOSMode` is `BETA`.
+When using `V2` mode, it is no longer necessary to set the `subBatchPerPartition` to `true`; it will default to `false` when the `EOSMode` is `V2`.
 
 Refer to https://cwiki.apache.org/confluence/display/KAFKA/KIP-447%3A+Producer+scalability+for+exactly+once+semantics[KIP-447] for more information.
 
+IMPORTANT: `kafka-clients` 3.0.0 and later no longer support `V2` (and automatic fallback to `V1`) with brokers earlier than 2.5; you must therefore override the default (`V2`) with `V1` if your brokers are older (or upgrade your brokers).
+
+`V1` and `V2` were previously `ALPHA` and `BETA`; they have been changed to align the framework with https://cwiki.apache.org/confluence/display/KAFKA/KIP-732%3A+Deprecate+eos-alpha+and+replace+eos-beta+with+eos-v2[KIP-732].
+
+
 [[interceptors]]
 ==== Wiring Spring Beans into Producer/Consumer Interceptors
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java

@@ -149,12 +149,49 @@ public enum EOSMode {
 		/**
 		 * 'transactional.id' fencing (0.11 - 2.4 brokers).
 		 */
-		ALPHA,
+		V1,
 
 		/**
 		 *  fetch-offset-request fencing (2.5+ brokers).
 		 */
-		BETA,
+		V2,
+
+		/**
+		 * 'transactional.id' fencing (0.11 - 2.4 brokers).
+		 * @deprecated in favor of {@link #V1}.
+		 */
+		@Deprecated
+		ALPHA(V1),
+
+		/**
+		 *  fetch-offset-request fencing (2.5+ brokers).
+		 *  @deprecated in favor of {@link #V2}.
+		 */
+		@Deprecated
+		BETA(V2);
+
+
+		final EOSMode mode;
+
+		EOSMode() {
+			this.mode = this;
+		}
+
+		/**
+		 * Create an alias.
+		 * @param v22 the mode for which this is an alias.
+		 */
+		EOSMode(EOSMode v12) {
+			this.mode = v12;
+		}
+
+		/**
+		 * Return the mode or the aliased mode.
+		 * @return the mode.
+		 */
+		public EOSMode getMode() {
+			return this.mode;
+		}
 
 	}
 
@@ -275,7 +312,7 @@ public enum EOSMode {
 
 	private boolean deliveryAttemptHeader;
 
-	private EOSMode eosMode = EOSMode.BETA;
+	private EOSMode eosMode = EOSMode.V2;
 
 	private TransactionDefinition transactionDefinition;
 
@@ -755,12 +792,14 @@ public EOSMode getEosMode() {
 	}
 
 	/**
-	 * Set the exactly once semantics mode. When {@link EOSMode#ALPHA} a producer per
+	 * Set the exactly once semantics mode. When {@link EOSMode#V1} a producer per
 	 * group/topic/partition is used (enabling 'transactional.id fencing`).
-	 * {@link EOSMode#BETA} enables fetch-offset-request fencing, and requires brokers 2.5
-	 * or later. With the 2.6 client, the default is now BETA because the 2.6 client can
+	 * {@link EOSMode#V2} enables fetch-offset-request fencing, and requires brokers 2.5
+	 * or later. With the 2.6 client, the default is now V2 because the 2.6 client can
 	 * automatically fall back to ALPHA.
-	 * @param eosMode the mode; default BETA.
+	 * IMPORTANT the 3.0 clients cannot be used with {@link EOSMode#V2} unless the broker
+	 * is 2.5 or higher.
+	 * @param eosMode the mode; default V2.
 	 * @since 2.5
 	 */
 	public void setEosMode(EOSMode eosMode) {

spring-kafka/src/main/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessor.java

@@ -137,7 +137,7 @@ public void process(List<ConsumerRecord<K, V>> records, Consumer<K, V> consumer,
 				getRecoveryStrategy((List) records, exception), container, this.logger)
 					&& isCommitRecovered() && this.kafkaTemplate.isTransactional()) {
 			ConsumerRecord<K, V> skipped = records.get(0);
-			if (EOSMode.ALPHA.equals(eosMode)) {
+			if (EOSMode.V1.equals(eosMode.getMode())) {
 				this.kafkaTemplate.sendOffsetsToTransaction(
 						Collections.singletonMap(new TopicPartition(skipped.topic(), skipped.partition()),
 								new OffsetAndMetadata(skipped.offset() + 1)));

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -905,7 +905,7 @@ private boolean setupSubBatchPerPartition() {
 			if (this.transactionManager == null) {
 				return false;
 			}
-			return this.eosMode.equals(EOSMode.ALPHA);
+			return this.eosMode.getMode().equals(EOSMode.V1);
 		}
 
 		@Nullable
@@ -2689,7 +2689,7 @@ private void sendOffsetsToTransaction() {
 		}
 
 		private void doSendOffsets(Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
-			if (this.eosMode.equals(EOSMode.ALPHA)) {
+			if (this.eosMode.getMode().equals(EOSMode.V1)) {
 				prod.sendOffsetsToTransaction(commits, this.consumerGroupId);
 			}
 			else {

spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultAfterRollbackProcessorTests.java

@@ -75,18 +75,18 @@ void testClassifier() {
 		Consumer<String, String> consumer = mock(Consumer.class);
 		given(consumer.groupMetadata()).willReturn(new ConsumerGroupMetadata("foo"));
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
-		processor.process(records, consumer, container, illegalState, true, EOSMode.ALPHA);
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V1);
 		processor.process(records, consumer, container,
-				new DeserializationException("intended", null, false, illegalState), true, EOSMode.ALPHA);
+				new DeserializationException("intended", null, false, illegalState), true, EOSMode.V1);
 		verify(template).sendOffsetsToTransaction(anyMap());
 		verify(template, never()).sendOffsetsToTransaction(anyMap(), any(ConsumerGroupMetadata.class));
 		assertThat(recovered.get()).isSameAs(record1);
 		processor.addNotRetryableExceptions(IllegalStateException.class);
 		recovered.set(null);
 		recovererShouldFail.set(true);
-		processor.process(records, consumer, container, illegalState, true, EOSMode.ALPHA);
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V1);
 		verify(template, times(1)).sendOffsetsToTransaction(anyMap()); // recovery failed
-		processor.process(records, consumer, container, illegalState, true, EOSMode.BETA);
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
 		verify(template, times(1)).sendOffsetsToTransaction(anyMap(), any(ConsumerGroupMetadata.class));
 		assertThat(recovered.get()).isSameAs(record1);
 		InOrder inOrder = inOrder(consumer);
@@ -123,12 +123,12 @@ void testBatchBackOff() {
 		given(consumer.groupMetadata()).willReturn(new ConsumerGroupMetadata("foo"));
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
 		given(container.isRunning()).willReturn(true);
-		processor.process(records, consumer, container, illegalState, false, EOSMode.BETA);
-		processor.process(records, consumer, container, illegalState, false, EOSMode.BETA);
+		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
+		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
 		verify(backOff, times(2)).start();
 		verify(execution.get(), times(2)).nextBackOff();
 		processor.clearThreadState();
-		processor.process(records, consumer, container, illegalState, false, EOSMode.BETA);
+		processor.process(records, consumer, container, illegalState, false, EOSMode.V2);
 		verify(backOff, times(3)).start();
 	}
 
@@ -144,7 +144,7 @@ void testEarlyExitBackOff() {
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
 		given(container.isRunning()).willReturn(false);
 		long t1 = System.currentTimeMillis();
-		processor.process(records, consumer, container, illegalState, true, EOSMode.BETA);
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
 		assertThat(System.currentTimeMillis() < t1 + 5_000);
 	}
 
@@ -161,7 +161,7 @@ void testNoEarlyExitBackOff() {
 		MessageListenerContainer container = mock(MessageListenerContainer.class);
 		given(container.isRunning()).willReturn(true);
 		long t1 = System.currentTimeMillis();
-		processor.process(records, consumer, container, illegalState, true, EOSMode.BETA);
+		processor.process(records, consumer, container, illegalState, true, EOSMode.V2);
 		assertThat(System.currentTimeMillis() >= t1 + 200);
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/listener/SubBatchPerPartitionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -156,7 +156,7 @@ void defaults() {
 		containerProps = new ContainerProperties("sbpp");
 		containerProps.setMessageListener(mock(MessageListener.class));
 		containerProps.setTransactionManager(mock(PlatformTransactionManager.class));
-		containerProps.setEosMode(EOSMode.ALPHA);
+		containerProps.setEosMode(EOSMode.V1);
 		container = new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();
 		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.subBatchPerPartition"))
@@ -166,7 +166,7 @@ void defaults() {
 		containerProps = new ContainerProperties("sbpp");
 		containerProps.setMessageListener(mock(MessageListener.class));
 		containerProps.setTransactionManager(mock(PlatformTransactionManager.class));
-		containerProps.setEosMode(EOSMode.BETA);
+		containerProps.setEosMode(EOSMode.V2);
 		container = new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.start();
 		assertThat(KafkaTestUtils.getPropertyValue(container, "listenerConsumer.subBatchPerPartition"))

spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java

@@ -146,27 +146,27 @@ public static void setup() {
 
 	@Test
 	public void testConsumeAndProduceTransactionKTM() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.ALPHA);
+		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.V1);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionHandleError() throws Exception {
-		testConsumeAndProduceTransactionGuts(true, AckMode.RECORD, EOSMode.ALPHA);
+		testConsumeAndProduceTransactionGuts(true, AckMode.RECORD, EOSMode.V1);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionKTMManual() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, AckMode.MANUAL_IMMEDIATE, EOSMode.ALPHA);
+		testConsumeAndProduceTransactionGuts(false, AckMode.MANUAL_IMMEDIATE, EOSMode.V1);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionKTM_BETA() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.BETA);
+		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.V1);
 	}
 
 	@Test
 	public void testConsumeAndProduceTransactionStopWhenFenced() throws Exception {
-		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.BETA, true);
+		testConsumeAndProduceTransactionGuts(false, AckMode.RECORD, EOSMode.V2, true);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -272,7 +272,7 @@ private void testConsumeAndProduceTransactionGuts(boolean handleError, AckMode a
 		assertThat(closeLatch.await(10, TimeUnit.SECONDS)).isTrue();
 		InOrder inOrder = inOrder(producer);
 		inOrder.verify(producer).beginTransaction();
-		if (eosMode.equals(EOSMode.ALPHA)) {
+		if (eosMode.equals(EOSMode.V1)) {
 			inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 					new OffsetAndMetadata(0)), "group");
 		}
@@ -291,7 +291,7 @@ private void testConsumeAndProduceTransactionGuts(boolean handleError, AckMode a
 			ArgumentCaptor<ProducerRecord> captor = ArgumentCaptor.forClass(ProducerRecord.class);
 			inOrder.verify(producer).send(captor.capture(), any(Callback.class));
 			assertThat(captor.getValue()).isEqualTo(new ProducerRecord("bar", "baz"));
-			if (eosMode.equals(EOSMode.ALPHA)) {
+			if (eosMode.equals(EOSMode.V1)) {
 				inOrder.verify(producer).sendOffsetsToTransaction(Collections.singletonMap(topicPartition,
 						new OffsetAndMetadata(1)), "group");
 			}