Introduce share consumer factories for Kafka Queues (Early Access)
* Introduce share consumer factories for Kafka Queues (Early Access) - Preliminary set of changes to support Kafka queueus introduced via KIP-932 (Kafka Queue) for early access in Apache Kafka 4.0.0. See: https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka - Add ShareConsumerFactory interface and DefaultShareConsumerFactory implementation as the initial building blocks for supporting Kafka Queues (KIP-932) in Spring for Apache Kafka 4.0.x. This factory and the implementation provide a flexible API for creating share consumers, and are designed as the foundation for further queue integration. - Tests to verify the share consumer behavior Related to #3875 https://github.com/spring-projects/spring-kafka/issues/3875 Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>
This commit is contained in:
Soby Chacko
@@ -0,0 +1,355 @@
|
||||
/*
|
||||
* Copyright 2025-2025 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.core;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.apache.kafka.clients.consumer.KafkaShareConsumer;
|
||||
import org.apache.kafka.clients.consumer.ShareConsumer;
|
||||
import org.apache.kafka.common.MetricName;
|
||||
import org.apache.kafka.common.serialization.Deserializer;
|
||||
import org.jspecify.annotations.Nullable;
|
||||
|
||||
import org.springframework.beans.factory.BeanNameAware;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* The {@link ShareConsumerFactory} implementation to produce new {@link ShareConsumer} instances
|
||||
* for provided {@link Map} {@code configs} and optional {@link Deserializer}s on each
|
||||
* {@link #createShareConsumer(String, String)} invocation.
|
||||
* <p>
|
||||
* If you are using {@link Deserializer}s that have no-arg constructors and require no setup, then simplest to
|
||||
* specify {@link Deserializer} classes in the configs passed to the
|
||||
* {@link DefaultShareConsumerFactory} constructor.
|
||||
* <p>
|
||||
* If that is not possible, but you are using {@link Deserializer}s that may be shared between all {@link ShareConsumer}
|
||||
* instances (and specifically that their close() method is a no-op), then you can pass in {@link Deserializer}
|
||||
* instances for one or both of the key and value deserializers.
|
||||
* <p>
|
||||
* If neither of the above is true then you may provide a {@link Supplier} for one or both {@link Deserializer}s
|
||||
* which will be used to obtain {@link Deserializer}(s) each time a {@link ShareConsumer} is created by the factory.
|
||||
*
|
||||
* @param <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
* @author Soby Chacko
|
||||
* @since 4.0
|
||||
*/
|
||||
public class DefaultShareConsumerFactory<K, V> extends KafkaResourceFactory
|
||||
implements ShareConsumerFactory<K, V>, BeanNameAware {
|
||||
|
||||
private final Map<String, Object> configs;
|
||||
|
||||
private @Nullable Supplier<@Nullable Deserializer<K>> keyDeserializerSupplier;
|
||||
|
||||
private @Nullable Supplier<@Nullable Deserializer<V>> valueDeserializerSupplier;
|
||||
|
||||
private boolean configureDeserializers = true;
|
||||
|
||||
private final List<Listener<K, V>> listeners = new ArrayList<>();
|
||||
|
||||
private String beanName = "not.managed.by.Spring";
|
||||
|
||||
/**
|
||||
* Construct a factory with the provided configuration.
|
||||
* @param configs the configuration.
|
||||
*/
|
||||
public DefaultShareConsumerFactory(Map<String, Object> configs) {
|
||||
this(configs, null, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct a factory with the provided configuration and deserializer suppliers.
|
||||
* When the suppliers are invoked to get an instance, the deserializers'
|
||||
* {@code configure()} methods will be called with the configuration map.
|
||||
* @param configs the configuration.
|
||||
* @param keyDeserializerSupplier the key {@link Deserializer} supplier function (nullable).
|
||||
* @param valueDeserializerSupplier the value {@link Deserializer} supplier function (nullable).
|
||||
*/
|
||||
public DefaultShareConsumerFactory(Map<String, Object> configs,
|
||||
@Nullable Supplier<@Nullable Deserializer<K>> keyDeserializerSupplier,
|
||||
@Nullable Supplier<@Nullable Deserializer<V>> valueDeserializerSupplier) {
|
||||
this(configs, keyDeserializerSupplier, valueDeserializerSupplier, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct a factory with the provided configuration and deserializers.
|
||||
* The deserializers' {@code configure()} methods will be called with the
|
||||
* configuration map unless {@code configureDeserializers} is false.
|
||||
* @param configs the configuration.
|
||||
* @param keyDeserializer the key {@link Deserializer}.
|
||||
* @param valueDeserializer the value {@link Deserializer}.
|
||||
* @param configureDeserializers false to not configure the deserializers.
|
||||
*/
|
||||
public DefaultShareConsumerFactory(Map<String, Object> configs,
|
||||
@Nullable Deserializer<K> keyDeserializer,
|
||||
@Nullable Deserializer<V> valueDeserializer, boolean configureDeserializers) {
|
||||
this(configs, keyDeserializer != null ? () -> keyDeserializer : null,
|
||||
valueDeserializer != null ? () -> valueDeserializer : null, configureDeserializers);
|
||||
}
|
||||
|
||||
/**
|
||||
* Construct a factory with the provided configuration, deserializer suppliers, and deserializer config flag.
|
||||
* When the suppliers are invoked to get an instance, the deserializers'
|
||||
* {@code configure()} methods will be called with the configuration map unless
|
||||
* {@code configureDeserializers} is false.
|
||||
* @param configs the configuration.
|
||||
* @param keyDeserializerSupplier the key {@link Deserializer} supplier function (nullable).
|
||||
* @param valueDeserializerSupplier the value {@link Deserializer} supplier function (nullable).
|
||||
* @param configureDeserializers whether to configure deserializers.
|
||||
*/
|
||||
public DefaultShareConsumerFactory(Map<String, Object> configs,
|
||||
@Nullable Supplier<@Nullable Deserializer<K>> keyDeserializerSupplier,
|
||||
@Nullable Supplier<@Nullable Deserializer<V>> valueDeserializerSupplier,
|
||||
boolean configureDeserializers) {
|
||||
this.configs = new ConcurrentHashMap<>(configs);
|
||||
this.configureDeserializers = configureDeserializers;
|
||||
this.keyDeserializerSupplier = keyDeserializerSupplier;
|
||||
this.valueDeserializerSupplier = valueDeserializerSupplier;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a share consumer with the provided group id and client id.
|
||||
* @param groupId the group id (maybe null).
|
||||
* @param clientId the client id.
|
||||
* @return the share consumer.
|
||||
*/
|
||||
@Override
|
||||
public ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId) {
|
||||
return createRawConsumer(groupId, clientId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Actually create the consumer.
|
||||
* @param groupId the group id (maybe null).
|
||||
* @param clientId the client id.
|
||||
* @return the share consumer.
|
||||
*/
|
||||
protected ShareConsumer<K, V> createRawConsumer(@Nullable String groupId, @Nullable String clientId) {
|
||||
Map<String, Object> consumerProperties = new HashMap<>(this.configs);
|
||||
if (groupId != null) {
|
||||
consumerProperties.put("group.id", groupId);
|
||||
}
|
||||
if (clientId != null) {
|
||||
consumerProperties.put("client.id", clientId);
|
||||
}
|
||||
return new ExtendedShareConsumer(consumerProperties);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanName(String name) {
|
||||
this.beanName = name;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the key deserializer. The deserializer will be configured using the consumer
|
||||
* configuration, unless {@link #setConfigureDeserializers(boolean)
|
||||
* configureDeserializers} is false.
|
||||
* @param keyDeserializer the deserializer.
|
||||
*/
|
||||
public void setKeyDeserializer(@Nullable Deserializer<K> keyDeserializer) {
|
||||
this.keyDeserializerSupplier = () -> keyDeserializer;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the value deserializer. The deserializer will be configured using the consumer
|
||||
* configuration, unless {@link #setConfigureDeserializers(boolean)
|
||||
* configureDeserializers} is false.
|
||||
* @param valueDeserializer the value deserializer.
|
||||
*/
|
||||
public void setValueDeserializer(@Nullable Deserializer<V> valueDeserializer) {
|
||||
this.valueDeserializerSupplier = () -> valueDeserializer;
|
||||
}
|
||||
|
||||
@Override
|
||||
@Nullable
|
||||
public Deserializer<K> getKeyDeserializer() {
|
||||
return this.keyDeserializerSupplier != null ? this.keyDeserializerSupplier.get() : null;
|
||||
}
|
||||
|
||||
@Override
|
||||
@Nullable
|
||||
public Deserializer<V> getValueDeserializer() {
|
||||
return this.valueDeserializerSupplier != null ? this.valueDeserializerSupplier.get() : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a supplier to supply instances of the key deserializer. The deserializer will
|
||||
* be configured using the consumer configuration, unless
|
||||
* {@link #setConfigureDeserializers(boolean) configureDeserializers} is false.
|
||||
* @param keyDeserializerSupplier the supplier (nullable).
|
||||
*/
|
||||
public void setKeyDeserializerSupplier(@Nullable Supplier<@Nullable Deserializer<K>> keyDeserializerSupplier) {
|
||||
this.keyDeserializerSupplier = keyDeserializerSupplier;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set a supplier to supply instances of the value deserializer. The deserializer will
|
||||
* be configured using the consumer configuration, unless
|
||||
* {@link #setConfigureDeserializers(boolean) configureDeserializers} is false.
|
||||
* @param valueDeserializerSupplier the supplier (nullable).
|
||||
*/
|
||||
public void setValueDeserializerSupplier(@Nullable Supplier<@Nullable Deserializer<V>> valueDeserializerSupplier) {
|
||||
this.valueDeserializerSupplier = valueDeserializerSupplier;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set to false (default true) to prevent programmatically provided deserializers (via
|
||||
* constructor or setters) from being configured using the consumer configuration,
|
||||
* e.g. if the deserializers are already fully configured.
|
||||
* @param configureDeserializers false to not configure.
|
||||
* @see #setKeyDeserializer(Deserializer)
|
||||
* @see #setKeyDeserializerSupplier(Supplier)
|
||||
* @see #setValueDeserializer(Deserializer)
|
||||
* @see #setValueDeserializerSupplier(Supplier)
|
||||
**/
|
||||
public void setConfigureDeserializers(boolean configureDeserializers) {
|
||||
this.configureDeserializers = configureDeserializers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current list of listeners.
|
||||
* @return the listeners.
|
||||
*/
|
||||
@Override
|
||||
public List<Listener<K, V>> getListeners() {
|
||||
return Collections.unmodifiableList(this.listeners);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a listener.
|
||||
* @param listener the listener.
|
||||
*/
|
||||
@Override
|
||||
public void addListener(Listener<K, V> listener) {
|
||||
Assert.notNull(listener, "'listener' cannot be null");
|
||||
this.listeners.add(listener);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a listener at a specific index.
|
||||
* <p>
|
||||
* This method allows insertion of a listener at a particular position in the internal listener list.
|
||||
* While this enables ordering of listener callbacks (which can be important for certain monitoring or extension scenarios),
|
||||
* there is intentionally no corresponding {@code removeListener(int index)} contract. Removing listeners by index is
|
||||
* discouraged because the position of a listener can change if others are added or removed, making it easy to
|
||||
* accidentally remove the wrong one. Managing listeners by their reference (object) is safer and less error-prone,
|
||||
* especially as listeners are usually set up once during initialization.
|
||||
* {@see #removeListener(Listener)}
|
||||
* </p>
|
||||
* @param index the index (list position).
|
||||
* @param listener the listener to add.
|
||||
*/
|
||||
@Override
|
||||
public void addListener(int index, Listener<K, V> listener) {
|
||||
Assert.notNull(listener, "'listener' cannot be null");
|
||||
if (index >= this.listeners.size()) {
|
||||
this.listeners.add(listener);
|
||||
}
|
||||
else {
|
||||
this.listeners.add(index, listener);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a listener.
|
||||
* @param listener the listener.
|
||||
* @return true if removed.
|
||||
*/
|
||||
@Override
|
||||
public boolean removeListener(Listener<K, V> listener) {
|
||||
return this.listeners.remove(listener);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
private Deserializer<K> keyDeserializer(Map<String, Object> configs) {
|
||||
Deserializer<K> deserializer =
|
||||
this.keyDeserializerSupplier != null
|
||||
? this.keyDeserializerSupplier.get()
|
||||
: null;
|
||||
if (deserializer != null && this.configureDeserializers) {
|
||||
deserializer.configure(configs, true);
|
||||
}
|
||||
return deserializer;
|
||||
}
|
||||
|
||||
@Nullable
|
||||
private Deserializer<V> valueDeserializer(Map<String, Object> configs) {
|
||||
Deserializer<V> deserializer =
|
||||
this.valueDeserializerSupplier != null
|
||||
? this.valueDeserializerSupplier.get()
|
||||
: null;
|
||||
if (deserializer != null && this.configureDeserializers) {
|
||||
deserializer.configure(configs, false);
|
||||
}
|
||||
return deserializer;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Object> getConfigurationProperties() {
|
||||
return Collections.unmodifiableMap(this.configs);
|
||||
}
|
||||
|
||||
protected class ExtendedShareConsumer extends KafkaShareConsumer<K, V> {
|
||||
|
||||
private @Nullable String idForListeners;
|
||||
|
||||
protected ExtendedShareConsumer(Map<String, Object> configProps) {
|
||||
super(configProps, keyDeserializer(configProps), valueDeserializer(configProps));
|
||||
|
||||
if (!DefaultShareConsumerFactory.this.listeners.isEmpty()) {
|
||||
Iterator<MetricName> metricIterator = metrics().keySet().iterator();
|
||||
String clientId = "unknown";
|
||||
if (metricIterator.hasNext()) {
|
||||
clientId = metricIterator.next().tags().get("client-id");
|
||||
}
|
||||
this.idForListeners = DefaultShareConsumerFactory.this.beanName + "." + clientId;
|
||||
for (Listener<K, V> listener : DefaultShareConsumerFactory.this.listeners) {
|
||||
listener.consumerAdded(this.idForListeners, this);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
super.close();
|
||||
notifyConsumerRemoved();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close(Duration timeout) {
|
||||
super.close(timeout);
|
||||
notifyConsumerRemoved();
|
||||
}
|
||||
|
||||
private void notifyConsumerRemoved() {
|
||||
for (Listener<K, V> listener : DefaultShareConsumerFactory.this.listeners) {
|
||||
listener.consumerRemoved(this.idForListeners, this);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,132 @@
|
||||
/*
|
||||
* Copyright 2025-2025 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.core;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.kafka.clients.consumer.ShareConsumer;
|
||||
import org.apache.kafka.common.serialization.Deserializer;
|
||||
import org.jspecify.annotations.Nullable;
|
||||
|
||||
/**
|
||||
* The strategy to produce a {@link ShareConsumer} instance for Kafka queue support.
|
||||
*
|
||||
* @param <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*
|
||||
* @author Soby Chacko
|
||||
* @since 4.0
|
||||
*/
|
||||
public interface ShareConsumerFactory<K, V> {
|
||||
|
||||
/**
|
||||
* Create a share consumer with the provided group id and client id.
|
||||
* @param groupId the group id (maybe null).
|
||||
* @param clientId the client id.
|
||||
* @return the share consumer.
|
||||
*/
|
||||
ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId);
|
||||
|
||||
/**
|
||||
* Return an unmodifiable reference to the configuration map for this factory.
|
||||
* Useful for cloning to make a similar factory.
|
||||
* @return the configs.
|
||||
*/
|
||||
default Map<String, Object> getConfigurationProperties() {
|
||||
throw new UnsupportedOperationException("'getConfigurationProperties()' is not supported");
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the configured key deserializer (if provided as an object instead
|
||||
* of a class name in the properties).
|
||||
* @return the deserializer.
|
||||
*/
|
||||
@Nullable
|
||||
default Deserializer<K> getKeyDeserializer() {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the configured value deserializer (if provided as an object instead
|
||||
* of a class name in the properties).
|
||||
* @return the deserializer.
|
||||
*/
|
||||
@Nullable
|
||||
default Deserializer<V> getValueDeserializer() {
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a listener.
|
||||
* @param listener the listener.
|
||||
* @return true if removed.
|
||||
*/
|
||||
default boolean removeListener(Listener<K, V> listener) {
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a listener at a specific index.
|
||||
* @param index the index (list position).
|
||||
* @param listener the listener.
|
||||
*/
|
||||
default void addListener(int index, Listener<K, V> listener) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a listener.
|
||||
* @param listener the listener.
|
||||
*/
|
||||
default void addListener(Listener<K, V> listener) {
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current list of listeners.
|
||||
* @return the listeners.
|
||||
*/
|
||||
default List<Listener<K, V>> getListeners() {
|
||||
return Collections.emptyList();
|
||||
}
|
||||
|
||||
/**
|
||||
* Listener for share consumer lifecycle events.
|
||||
*
|
||||
* @param <K> the key type.
|
||||
* @param <V> the value type.
|
||||
*/
|
||||
interface Listener<K, V> {
|
||||
|
||||
/**
|
||||
* A new consumer was created.
|
||||
* @param id the consumer id (factory bean name and client.id separated by a period).
|
||||
* @param consumer the consumer.
|
||||
*/
|
||||
default void consumerAdded(String id, ShareConsumer<K, V> consumer) {
|
||||
}
|
||||
|
||||
/**
|
||||
* An existing consumer was removed.
|
||||
* @param id the consumer id (factory bean name and client.id separated by a period).
|
||||
* @param consumer the consumer.
|
||||
*/
|
||||
default void consumerRemoved(@Nullable String id, ShareConsumer<K, V> consumer) {
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,200 @@
|
||||
/*
|
||||
* Copyright 2025-2025 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.core;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.apache.kafka.clients.admin.Admin;
|
||||
import org.apache.kafka.clients.admin.AdminClient;
|
||||
import org.apache.kafka.clients.admin.AlterConfigOp;
|
||||
import org.apache.kafka.clients.admin.ConfigEntry;
|
||||
import org.apache.kafka.clients.consumer.ShareConsumer;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.apache.kafka.common.config.ConfigResource;
|
||||
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||
import org.apache.kafka.common.serialization.StringSerializer;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.kafka.test.EmbeddedKafkaBroker;
|
||||
import org.springframework.kafka.test.context.EmbeddedKafka;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* Basic tests for {@link DefaultShareConsumerFactory}.
|
||||
*
|
||||
* @author Soby Chacko
|
||||
* @since 4.0
|
||||
*/
|
||||
@EmbeddedKafka(
|
||||
topics = {"embedded-share-test"}, partitions = 1,
|
||||
brokerProperties = {
|
||||
"unstable.api.versions.enable=true",
|
||||
"group.coordinator.rebalance.protocols=classic,share",
|
||||
"share.coordinator.state.topic.replication.factor=1",
|
||||
"share.coordinator.state.topic.min.isr=1"
|
||||
})
|
||||
class DefaultShareConsumerFactoryTests {
|
||||
|
||||
@Test
|
||||
void shouldInstantiateWithConfigs() {
|
||||
Map<String, Object> configs = new HashMap<>();
|
||||
configs.put("bootstrap.servers", "localhost:9092");
|
||||
DefaultShareConsumerFactory<String, String> factory = new DefaultShareConsumerFactory<>(configs);
|
||||
assertThat(factory).isNotNull();
|
||||
assertThat(factory.getConfigurationProperties()).containsKey("bootstrap.servers");
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldConfigureDeserializersViaSetters() {
|
||||
var configs = new HashMap<String, Object>();
|
||||
var factory = new DefaultShareConsumerFactory<String, String>(configs);
|
||||
var keyDeserializer = new StringDeserializer();
|
||||
var valueDeserializer = new StringDeserializer();
|
||||
factory.setKeyDeserializer(keyDeserializer);
|
||||
factory.setValueDeserializer(valueDeserializer);
|
||||
assertThat(factory.getKeyDeserializer())
|
||||
.as("Key deserializer should match the set instance")
|
||||
.isSameAs(keyDeserializer);
|
||||
assertThat(factory.getValueDeserializer())
|
||||
.as("Value deserializer should match the set instance")
|
||||
.isSameAs(valueDeserializer);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldConfigureDeserializersViaConstructor() {
|
||||
var configs = new HashMap<String, Object>();
|
||||
var keyDeserializer = new StringDeserializer();
|
||||
var valueDeserializer = new StringDeserializer();
|
||||
var factory = new DefaultShareConsumerFactory<>(configs, keyDeserializer, valueDeserializer, true);
|
||||
assertThat(factory.getKeyDeserializer())
|
||||
.as("Key deserializer should match the constructor instance")
|
||||
.isSameAs(keyDeserializer);
|
||||
assertThat(factory.getValueDeserializer())
|
||||
.as("Value deserializer should match the constructor instance")
|
||||
.isSameAs(valueDeserializer);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldRegisterAndRemoveListeners() {
|
||||
var configs = new HashMap<String, Object>();
|
||||
var factory = new DefaultShareConsumerFactory<String, String>(configs);
|
||||
var listener = new ShareConsumerFactory.Listener<String, String>() {
|
||||
|
||||
};
|
||||
factory.addListener(listener);
|
||||
assertThat(factory.getListeners())
|
||||
.as("Listeners should contain the added listener")
|
||||
.contains(listener);
|
||||
factory.removeListener(listener);
|
||||
assertThat(factory.getListeners())
|
||||
.as("Listeners should not contain the removed listener")
|
||||
.doesNotContain(listener);
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldCreateShareConsumer() {
|
||||
Map<String, Object> configs = new HashMap<>();
|
||||
configs.put("bootstrap.servers", "localhost:9092");
|
||||
configs.put("key.deserializer", StringDeserializer.class);
|
||||
configs.put("value.deserializer", StringDeserializer.class);
|
||||
DefaultShareConsumerFactory<String, String> factory = new DefaultShareConsumerFactory<>(configs);
|
||||
ShareConsumer<String, String> shareConsumer = factory.createShareConsumer("group", "myapp-client-id");
|
||||
assertThat(shareConsumer).isNotNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void shouldReturnUnmodifiableListenersList() {
|
||||
var configs = new HashMap<String, Object>();
|
||||
var factory = new DefaultShareConsumerFactory<String, String>(configs);
|
||||
var listener = new ShareConsumerFactory.Listener<String, String>() {
|
||||
|
||||
};
|
||||
factory.addListener(listener);
|
||||
var listeners = factory.getListeners();
|
||||
assertThat(listeners).contains(listener);
|
||||
// Attempting to modify the returned list should throw
|
||||
assertThatThrownBy(() -> listeners.add(new ShareConsumerFactory.Listener<>() {
|
||||
|
||||
}))
|
||||
.as("Listeners list should be unmodifiable")
|
||||
.isInstanceOf(UnsupportedOperationException.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
@SuppressWarnings("try")
|
||||
void integrationTestDefaultShareConsumerFactory(EmbeddedKafkaBroker broker) throws Exception {
|
||||
final String topic = "embedded-share-test";
|
||||
final String groupId = "testGroup";
|
||||
var bootstrapServers = broker.getBrokersAsString();
|
||||
|
||||
var producerProps = new java.util.Properties();
|
||||
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
||||
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
|
||||
|
||||
try (var producer = new KafkaProducer<String, String>(producerProps)) {
|
||||
producer.send(new ProducerRecord<>(topic, "key", "integration-test-value")).get();
|
||||
}
|
||||
|
||||
Map<String, Object> adminProperties = new HashMap<>();
|
||||
adminProperties.put("bootstrap.servers", bootstrapServers);
|
||||
|
||||
// For this test: force new share groups to start from the beginning of the topic.
|
||||
// This is NOT the same as the usual consumer auto.offset.reset; it's a group config,
|
||||
// so use AdminClient to set share.auto.offset.reset = earliest for our test group.
|
||||
try (AdminClient ignored = AdminClient.create(adminProperties)) {
|
||||
ConfigEntry entry = new ConfigEntry("share.auto.offset.reset", "earliest");
|
||||
AlterConfigOp op = new AlterConfigOp(entry, AlterConfigOp.OpType.SET);
|
||||
|
||||
Map<ConfigResource, Collection<AlterConfigOp>> configs = Map.of(
|
||||
new ConfigResource(ConfigResource.Type.GROUP, "testGroup"), Arrays.asList(op));
|
||||
|
||||
try (Admin admin = AdminClient.create(adminProperties)) {
|
||||
admin.incrementalAlterConfigs(configs).all().get();
|
||||
}
|
||||
}
|
||||
|
||||
var consumerProps = new HashMap<String, Object>();
|
||||
consumerProps.put("bootstrap.servers", bootstrapServers);
|
||||
consumerProps.put("key.deserializer", StringDeserializer.class);
|
||||
consumerProps.put("value.deserializer", StringDeserializer.class);
|
||||
consumerProps.put("group.id", groupId);
|
||||
|
||||
var factory = new DefaultShareConsumerFactory<String, String>(consumerProps);
|
||||
var consumer = factory.createShareConsumer(groupId, "myapp-client-id");
|
||||
consumer.subscribe(Collections.singletonList(topic));
|
||||
|
||||
var records = consumer.poll(Duration.ofSeconds(10));
|
||||
assertThat(records.count())
|
||||
.as("Should have received at least one record")
|
||||
.isGreaterThan(0);
|
||||
assertThat(records.iterator().next().value())
|
||||
.as("Record value should match")
|
||||
.isEqualTo("integration-test-value");
|
||||
consumer.close();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user