GH-2381 : Support kafka parallel-consumer

build.gradle

@@ -73,6 +73,7 @@ ext {
 	springRetryVersion = '2.0.5'
 	springVersion = '6.1.5'
 	zookeeperVersion = '3.8.4'
+	parallelConsumerVersion = '0.5.2.8'
 
 	idPrefix = 'kafka'
 
@@ -289,6 +290,9 @@ project ('spring-kafka') {
 			exclude group: 'org.jetbrains.kotlin'
 		}
 
+		// Parallel Consumer
+		api "io.confluent.parallelconsumer:parallel-consumer-core:$parallelConsumerVersion"
+
 		// Spring Data projection message binding support
 		optionalApi ("org.springframework.data:spring-data-commons") {
 			exclude group: 'org.springframework'

spring-kafka/src/main/java/org/springframework/kafka/annotation/EnableParallelConsumer.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2014-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.annotation;
+
+import org.springframework.context.annotation.Import;
+import org.springframework.kafka.config.ParallelConsumerConfiguration;
+import org.springframework.kafka.config.ParallelConsumerImportSelector;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * If you want to import {@link ParallelConsumerConfiguration} to your application,
+ * you just annotated {@link EnableParallelConsumer} to your spring application.
+ *
+ * @author Sanghyeok An
+ *
+ * @since 3.3
+ */
+
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Import(ParallelConsumerImportSelector.class)
+public @interface EnableParallelConsumer {
+}

spring-kafka/src/main/java/org/springframework/kafka/config/OnMissingParallelConsumerOptionsProviderCondition.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Condition;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+import org.springframework.kafka.core.parallelconsumer.ParallelConsumerOptionsProvider;
+
+/**
+ * This class is to provide {@link ParallelConsumerOptionsProvider} as default.
+ * {@link ParallelConsumerConfiguration} use this function when {@link ApplicationContext} started.
+ * If there is no spring bean in {@link ApplicationContext} such as {@link ParallelConsumerOptionsProvider},
+ * {@link ParallelConsumerConfiguration} will register default {@link ParallelConsumerOptionsProvider}.
+ *
+ * @author Sanghyeok An
+ *
+ * @since 3.3
+ */
+public class OnMissingParallelConsumerOptionsProviderCondition implements Condition {
+
+	@Override
+	public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
+		return context.getBeanFactory().getBean(ParallelConsumerOptionsProvider.class) == null;
+	}
+}

spring-kafka/src/main/java/org/springframework/kafka/config/ParallelConsumerConfig.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright 2014-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+
+import org.apache.kafka.clients.consumer.Consumer;
+
+import io.confluent.parallelconsumer.ParallelConsumerOptions;
+
+import org.apache.kafka.clients.producer.Producer;
+import org.springframework.kafka.core.parallelconsumer.ParallelConsumerOptionsProvider;
+import org.springframework.kafka.annotation.EnableParallelConsumer;
+
+// It would be better to be migrated to org.springframework.boot.autoconfigure.kafka.KafkaProperties.
+
+/**
+ * ParallelConsumerConfig is for config of {@link io.confluent.parallelconsumer}.
+ * This will be registered as Spring Bean when {@link EnableParallelConsumer} is annotated to your spring application.
+ *
+ * @author Sanghyeok An
+ *
+ * @since 3.3
+ */
+
+public class ParallelConsumerConfig<K, V> {
+
+	public static final String DEFAULT_BEAN_NAME = "parallelConsumerConfig";
+	private final ParallelConsumerOptionsProvider<K, V> provider;
+
+	public ParallelConsumerConfig(ParallelConsumerOptionsProvider<K, V> provider) {
+		this.provider = provider;
+	}
+
+	public ParallelConsumerOptions<K, V> toConsumerOptions(
+			ParallelConsumerOptions.ParallelConsumerOptionsBuilder<K, V> builder,
+			Consumer<K, V> consumer,
+			Producer<K, V> producer) {
+
+		builder.producer(producer);
+		return toConsumerOptions(builder, consumer);
+	}
+
+	public ParallelConsumerOptions<K, V> toConsumerOptions(
+			ParallelConsumerOptions.ParallelConsumerOptionsBuilder<K, V> builder,
+			Consumer<K, V> consumer) {
+		builder.consumer(consumer);
+		return buildRemainOptions(builder);
+	}
+
+	private ParallelConsumerOptions<K, V> buildRemainOptions(ParallelConsumerOptions.ParallelConsumerOptionsBuilder<K, V> builder) {
+		if (this.provider.managedExecutorService() != null){
+			builder.managedExecutorService(this.provider.managedExecutorService());
+		}
+
+		if (this.provider.managedThreadFactory() != null){
+			builder.managedThreadFactory(this.provider.managedThreadFactory());
+		}
+
+		if (this.provider.meterRegistry() != null){
+			builder.meterRegistry(this.provider.meterRegistry());
+		}
+
+		if (this.provider.pcInstanceTag() != null){
+			builder.pcInstanceTag(this.provider.pcInstanceTag());
+		}
+
+		if (this.provider.metricsTags() != null){
+			builder.metricsTags(this.provider.metricsTags());
+		}
+
+		if (this.provider.allowEagerProcessingDuringTransactionCommit() != null){
+			builder.allowEagerProcessingDuringTransactionCommit(this.provider.allowEagerProcessingDuringTransactionCommit());
+		}
+
+		if (this.provider.commitLockAcquisitionTimeout() != null){
+			builder.commitLockAcquisitionTimeout(this.provider.commitLockAcquisitionTimeout());
+		}
+
+		if (this.provider.produceLockAcquisitionTimeout() != null){
+			builder.produceLockAcquisitionTimeout(this.provider.produceLockAcquisitionTimeout());
+		}
+
+		if (this.provider.commitInterval() != null){
+			builder.commitInterval(this.provider.commitInterval());
+		}
+
+		if (this.provider.ordering() != null){
+			builder.ordering(this.provider.ordering());
+		}
+
+		if (this.provider.commitMode() != null){
+			builder.commitMode(this.provider.commitMode());
+		}
+
+		if (this.provider.maxConcurrency() != null){
+			builder.maxConcurrency(this.provider.maxConcurrency());
+		}
+
+		if (this.provider.invalidOffsetMetadataPolicy() != null){
+			builder.invalidOffsetMetadataPolicy(this.provider.invalidOffsetMetadataPolicy());
+		}
+
+		if (this.provider.retryDelayProvider() != null){
+			builder.retryDelayProvider(this.provider.retryDelayProvider());
+		}
+
+		if (this.provider.sendTimeout() != null){
+			builder.sendTimeout(this.provider.sendTimeout());
+		}
+
+		if (this.provider.offsetCommitTimeout() != null){
+			builder.offsetCommitTimeout(this.provider.offsetCommitTimeout());
+		}
+
+		if (this.provider.batchSize() != null){
+			builder.batchSize(this.provider.batchSize());
+		}
+
+		if (this.provider.thresholdForTimeSpendInQueueWarning() != null){
+			builder.thresholdForTimeSpendInQueueWarning(this.provider.thresholdForTimeSpendInQueueWarning());
+		}
+
+		if (this.provider.maxFailureHistory() != null){
+			builder.maxFailureHistory(this.provider.maxFailureHistory());
+		}
+
+		if (this.provider.shutdownTimeout() != null){
+			builder.shutdownTimeout(this.provider.shutdownTimeout());
+		}
+
+		if (this.provider.drainTimeout() != null){
+			builder.drainTimeout(this.provider.drainTimeout());
+		}
+
+		if (this.provider.messageBufferSize() != null){
+			builder.messageBufferSize(this.provider.messageBufferSize());
+		}
+
+		if (this.provider.initialLoadFactor() != null){
+			builder.initialLoadFactor(this.provider.initialLoadFactor());
+		}
+		if (this.provider.maximumLoadFactor() != null){
+			builder.maximumLoadFactor(this.provider.maximumLoadFactor());
+		}
+
+		return builder.build();
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/config/ParallelConsumerConfiguration.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2014-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.parallelconsumer.ParallelConsumerRootInterface;
+import org.springframework.kafka.core.ParallelConsumerFactory;
+import org.springframework.kafka.annotation.EnableParallelConsumer;
+import org.springframework.kafka.core.parallelconsumer.ParallelConsumerOptionsProvider;
+
+/**
+ * If User decide to use parallelConsumer on SpringKafka, User should import this class to their ComponentScan scopes.
+ * If so, this class will register both {@link ParallelConsumerContext} and {@link ParallelConsumerFactory} as Spring Bean.
+ * User has responsibility
+ *   1. annotated {@link EnableParallelConsumer} on their spring application
+ *   2. register ConcreteClass of {@link ParallelConsumerRootInterface}.
+ *
+ * @author Sanghyoek An
+ *
+ * @since 3.3
+ */
+
+public class ParallelConsumerConfiguration<K, V> {
+
+	@Bean
+	@Conditional(OnMissingParallelConsumerOptionsProviderCondition.class)
+	public ParallelConsumerOptionsProvider<K, V> parallelConsumerOptionsProvider() {
+		return new ParallelConsumerOptionsProvider<K, V>() {};
+	}
+
+	@Bean(name = ParallelConsumerConfig.DEFAULT_BEAN_NAME)
+	public ParallelConsumerConfig<K, V> parallelConsumerConfig(ParallelConsumerOptionsProvider<K, V> parallelConsumerOptionsProvider) {
+		return new ParallelConsumerConfig<K, V>(parallelConsumerOptionsProvider);
+	}
+
+	@Bean(name = ParallelConsumerContext.DEFAULT_BEAN_NAME)
+	public ParallelConsumerContext<K,V> parallelConsumerContext(ParallelConsumerConfig<K, V> parallelConsumerConfig,
+																ParallelConsumerRootInterface<K, V> parallelConsumerCallback) {
+		return new ParallelConsumerContext<K, V>(parallelConsumerConfig,
+												 parallelConsumerCallback);
+	}
+
+	@Bean(name = ParallelConsumerFactory.DEFAULT_BEAN_NAME)
+	public ParallelConsumerFactory<K,V> parallelConsumerFactory(DefaultKafkaConsumerFactory<K,V> consumerFactory,
+																DefaultKafkaProducerFactory<K,V> producerFactory,
+																ParallelConsumerContext<K,V> parallelConsumerContext) {
+		return new ParallelConsumerFactory<K, V>(parallelConsumerContext, consumerFactory, producerFactory);
+	}
+
+}

spring-kafka/src/main/java/org/springframework/kafka/config/ParallelConsumerContext.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2014-2024 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.springframework.kafka.core.parallelconsumer.ParallelConsumerRootInterface;
+
+import io.confluent.parallelconsumer.ParallelConsumerOptions;
+import io.confluent.parallelconsumer.ParallelConsumerOptions.ParallelConsumerOptionsBuilder;
+
+/**
+ * This class is for collecting all related with ParallelConsumer.
+ *
+ * @author Sanghyeok An
+ *
+ * @since 3.3
+ */
+
+
+public class ParallelConsumerContext<K,V> {
+
+	public static final String DEFAULT_BEAN_NAME = "parallelConsumerContext";
+	private final ParallelConsumerConfig parallelConsumerConfig;
+	private final ParallelConsumerRootInterface<K, V> parallelConsumerCallback;
+
+	public ParallelConsumerContext(ParallelConsumerConfig<K, V> parallelConsumerConfig,
+								   ParallelConsumerRootInterface<K, V> callback) {
+		this.parallelConsumerConfig = parallelConsumerConfig;
+		this.parallelConsumerCallback = callback;
+	}
+
+	public ParallelConsumerRootInterface<K, V> parallelConsumerCallback() {
+		return this.parallelConsumerCallback;
+	}
+
+	public ParallelConsumerOptions<K, V> getParallelConsumerOptions(Consumer<K, V> consumer) {
+		final ParallelConsumerOptionsBuilder<K, V> builder = ParallelConsumerOptions.builder();
+		return parallelConsumerConfig.toConsumerOptions(builder, consumer);
+	}
+
+	public ParallelConsumerOptions<K, V> getParallelConsumerOptions(Consumer<K, V> consumer, Producer<K, V> producer) {
+		final ParallelConsumerOptionsBuilder<K, V> builder = ParallelConsumerOptions.builder();
+		return parallelConsumerConfig.toConsumerOptions(builder, consumer, producer);
+	}
+
+}