Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-161: Implement Gateways

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/161

Initial commit - outbound

TODO: inbound, DSL, XML, Docs

Fix reply topic.

Polishing - don't leak internal headers; don't propagate outbound headers (topic etc.).

Polishing

Validate reply topic and partition against replying template.

DSL

Checkstyle fixes.

Add inbound gateway.

Inbound DSL

Polishing - PR comments.

Polishing - PR comments

More polishing

Increase test timeouts.

Increase test timeouts.

* Code style polishing
* Fix `KafkaInboundGatewaySpec.getComponentsToRegister()` to `get()`
on Specs
* Remove `getContainer()` in the `KafkaMessageListenerContainerSpec` in
favor of `get()` in the `IntegrationComponentSpec`
This commit is contained in:
Gary Russell
2018-02-02 17:26:40 -05:00
committed by Artem Bilan
parent 86d8373084
commit 2b292f2a7d
13 changed files with 1908 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java
View File

@@ -26,7 +26,9 @@ import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.GenericMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
/**
@@ -34,6 +36,7 @@ import org.springframework.kafka.support.TopicPartitionInitialOffset;
*
* @author Artem Bilan
* @author Nasko Vasilev
* @author Gary Russell
*
* @since 3.0
*/
@@ -47,8 +50,8 @@ public final class Kafka {
* @param <S> the {@link KafkaProducerMessageHandlerSpec} extension type.
* @return the KafkaProducerMessageHandlerSpec.
*/
public static <K, V, S extends KafkaProducerMessageHandlerSpec<K, V, S>> KafkaProducerMessageHandlerSpec<K, V, S>
outboundChannelAdapter(KafkaTemplate<K, V> kafkaTemplate) {
public static <K, V, S extends KafkaProducerMessageHandlerSpec<K, V, S>> KafkaProducerMessageHandlerSpec<K, V, S> outboundChannelAdapter(
KafkaTemplate<K, V> kafkaTemplate) {
return new KafkaProducerMessageHandlerSpec<>(kafkaTemplate);
}
@@ -61,8 +64,8 @@ public final class Kafka {
* @return the KafkaProducerMessageHandlerSpec.
* @see <a href="https://kafka.apache.org/documentation.html#producerconfigs">Kafka Producer Configs</a>
*/
public static <K, V> KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<K, V>
outboundChannelAdapter(ProducerFactory<K, V> producerFactory) {
public static <K, V> KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<K, V> outboundChannelAdapter(
ProducerFactory<K, V> producerFactory) {
return new KafkaProducerMessageHandlerSpec.KafkaProducerMessageHandlerTemplateSpec<>(producerFactory);
}
@@ -77,8 +80,8 @@ public final class Kafka {
* @return the spec.
* @since 3.0.1
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V>
inboundChannelAdapter(ConsumerFactory<K, V> consumerFactory, String... topics) {
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory, String... topics) {
return new KafkaInboundChannelAdapterSpec<>(consumerFactory, topics);
}
@@ -94,9 +97,9 @@ public final class Kafka {
* @return the spec.
* @since 3.0.1
*/
public static <K, V> KafkaInboundChannelAdapterSpec<K, V>
inboundChannelAdapter(ConsumerFactory<K, V> consumerFactory,
KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
public static <K, V> KafkaInboundChannelAdapterSpec<K, V> inboundChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
KafkaAckCallbackFactory<K, V> ackCallbackFactory, String... topics) {
return new KafkaInboundChannelAdapterSpec<>(consumerFactory, ackCallbackFactory, topics);
}
@@ -109,8 +112,7 @@ public final class Kafka {
* @param <S> the {@link KafkaMessageDrivenChannelAdapterSpec} extension type.
* @return the KafkaMessageDrivenChannelAdapterSpec.
*/
public static <K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>>
KafkaMessageDrivenChannelAdapterSpec<K, V, S> messageDrivenChannelAdapter(
public static <K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>> KafkaMessageDrivenChannelAdapterSpec<K, V, S> messageDrivenChannelAdapter(
AbstractMessageListenerContainer<K, V> listenerContainer) {
return messageDrivenChannelAdapter(listenerContainer, KafkaMessageDrivenChannelAdapter.ListenerMode.record);
@@ -125,8 +127,7 @@ public final class Kafka {
* @param <A> the {@link KafkaMessageDrivenChannelAdapterSpec} extension type.
* @return the KafkaMessageDrivenChannelAdapterSpec.
*/
public static <K, V, A extends KafkaMessageDrivenChannelAdapterSpec<K, V, A>>
KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
public static <K, V, A extends KafkaMessageDrivenChannelAdapterSpec<K, V, A>> KafkaMessageDrivenChannelAdapterSpec<K, V, A> messageDrivenChannelAdapter(
AbstractMessageListenerContainer<K, V> listenerContainer,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
@@ -143,8 +144,8 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties) {
return messageDrivenChannelAdapter(consumerFactory, containerProperties,
KafkaMessageDrivenChannelAdapter.ListenerMode.record);
@@ -161,12 +162,12 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
return messageDrivenChannelAdapter(
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
new KafkaMessageListenerContainerSpec<>(consumerFactory,
containerProperties), listenerMode);
}
@@ -180,8 +181,8 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
TopicPartitionInitialOffset... topicPartitions) {
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
@@ -199,13 +200,13 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
TopicPartitionInitialOffset... topicPartitions) {
return messageDrivenChannelAdapter(
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
new KafkaMessageListenerContainerSpec<>(consumerFactory,
topicPartitions), listenerMode);
}
@@ -219,8 +220,8 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, String... topics) {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory, String... topics) {
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
topics);
@@ -237,12 +238,12 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, String... topics) {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode,
String... topics) {
return messageDrivenChannelAdapter(
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
new KafkaMessageListenerContainerSpec<>(consumerFactory,
topics), listenerMode);
}
@@ -256,8 +257,8 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
return messageDrivenChannelAdapter(consumerFactory, KafkaMessageDrivenChannelAdapter.ListenerMode.record,
topicPattern);
@@ -274,20 +275,108 @@ public final class Kafka {
* @return the KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec.
*/
public static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
ConsumerFactory<K, V> consumerFactory,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode, Pattern topicPattern) {
return messageDrivenChannelAdapter(
new KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<>(consumerFactory,
new KafkaMessageListenerContainerSpec<>(consumerFactory,
topicPattern),
listenerMode);
}
/**
* Create an initial {@link KafkaProducerMessageHandlerSpec}.
* @param kafkaTemplate the {@link ReplyingKafkaTemplate} to use
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @param <S> the {@link KafkaOutboundGatewaySpec} extension type.
* @return the KafkaGatewayMessageHandlerSpec.
* @since 3.0.2
*/
public static <K, V, R, S extends KafkaOutboundGatewaySpec<K, V, R, S>> KafkaOutboundGatewaySpec<K, V, R, S> outboundGateway(
ReplyingKafkaTemplate<K, V, R> kafkaTemplate) {
return new KafkaOutboundGatewaySpec<>(kafkaTemplate);
}
/**
* Create an initial {@link KafkaProducerMessageHandlerSpec} with ProducerFactory.
* @param producerFactory the {@link ProducerFactory} Java 8 Lambda.
* @param replyContainer a listener container for replies.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the KafkaGatewayMessageHandlerSpec.
* @since 3.0.2
*/
public static <K, V, R> KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec<K, V, R> outboundGateway(
ProducerFactory<K, V> producerFactory, GenericMessageListenerContainer<K, R> replyContainer) {
return new KafkaOutboundGatewaySpec.KafkaGatewayMessageHandlerTemplateSpec<>(producerFactory,
replyContainer);
}
/**
* Create an initial {@link KafkaInboundGatewaySpec} with the provided container and
* template.
* @param container the container.
* @param template the template.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @param <S> the {@link KafkaInboundGatewaySpec} extension type.
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R, S extends KafkaInboundGatewaySpec<K, V, R, S>> KafkaInboundGatewaySpec<K, V, R, S> inboundGateway(
AbstractMessageListenerContainer<K, V> container, KafkaTemplate<K, R> template) {
return new KafkaInboundGatewaySpec<>(container, template);
}
/**
* Create an initial {@link KafkaInboundGatewaySpec} with the provided consumer factory,
* container properties and producer factory.
* @param consumerFactory the consumer factory.
* @param containerProperties the container properties.
* @param producerFactory the producer factory.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R> KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<K, V, R> inboundGateway(
ConsumerFactory<K, V> consumerFactory, ContainerProperties containerProperties,
ProducerFactory<K, R> producerFactory) {
return inboundGateway(
new KafkaMessageListenerContainerSpec<>(consumerFactory, containerProperties),
new KafkaTemplateSpec<>(producerFactory));
}
/**
* Create an initial {@link KafkaInboundGatewaySpec} with the provided container and
* template specs.
* @param containerSpec the container spec.
* @param templateSpec the template spec.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type (request).
* @param <R> the Kafka message value type (reply).
* @return the spec.
* @since 3.0.2
*/
public static <K, V, R> KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<K, V, R> inboundGateway(
KafkaMessageListenerContainerSpec<K, V> containerSpec, KafkaTemplateSpec<K, R> templateSpec) {
return new KafkaInboundGatewaySpec.KafkaInboundGatewayListenerContainerSpec<>(containerSpec, templateSpec);
}
private static <K, V>
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V>
messageDrivenChannelAdapter(KafkaMessageDrivenChannelAdapterSpec.KafkaMessageListenerContainerSpec<K, V> spec,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
KafkaMessageDrivenChannelAdapterSpec.KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> messageDrivenChannelAdapter(
KafkaMessageListenerContainerSpec<K, V> spec, KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
return new KafkaMessageDrivenChannelAdapterSpec
.KafkaMessageDrivenChannelAdapterListenerContainerSpec<>(spec, listenerMode);

150
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java Normal file
View File

@@ -0,0 +1,150 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.dsl;
import java.util.Map;
import java.util.function.Consumer;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.MessagingGatewaySpec;
import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
import org.springframework.integration.support.ObjectStringMapBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
/**
* A {@link MessagingGatewaySpec} implementation for the {@link KafkaInboundGateway}.
*
* @param <K> the key type.
* @param <V> the request value type.
* @param <R> the reply value type.
* @param <S> the target {@link KafkaInboundGatewaySpec} implementation type.
*
* @author Gary Russell
*
* @since 3.0.2
*/
public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<K, V, R, S>>
extends MessagingGatewaySpec<S, KafkaInboundGateway<K, V, R>> {
KafkaInboundGatewaySpec(AbstractMessageListenerContainer<K, V> messageListenerContainer,
KafkaTemplate<K, R> kafkaTemplate) {
super(new KafkaInboundGateway<>(messageListenerContainer, kafkaTemplate));
}
/**
* Set the message converter to use with a record-based consumer.
* @param messageConverter the converter.
* @return the spec
*/
public S messageConverter(RecordMessageConverter messageConverter) {
this.target.setMessageConverter(messageConverter);
return _this();
}
/**
* Specify a {@link RetryTemplate} instance to wrap
* {@code KafkaInboundGateway.IntegrationRecordMessageListener} into
* {@link org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter}.
* @param retryTemplate the {@link RetryTemplate} to use.
* @return the spec
*/
public S retryTemplate(RetryTemplate retryTemplate) {
this.target.setRetryTemplate(retryTemplate);
return _this();
}
/**
* A {@link RecoveryCallback} instance for retry operation;
* if null, the exception will be thrown to the container after retries are exhausted.
* Does not make sense if {@link #retryTemplate(RetryTemplate)} isn't specified.
* @param recoveryCallback the recovery callback.
* @return the spec
*/
public S recoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.target.setRecoveryCallback(recoveryCallback);
return _this();
}
/**
* A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaInboundGatewaySpec}
* extension.
* @param <K> the key type.
* @param <V> the request value type.
* @param <R> the reply value type.
*/
public static class KafkaInboundGatewayListenerContainerSpec<K, V, R> extends
KafkaInboundGatewaySpec<K, V, R, KafkaInboundGatewayListenerContainerSpec<K, V, R>>
implements ComponentsRegistration {
private final KafkaMessageListenerContainerSpec<K, V> containerSpec;
private final KafkaTemplateSpec<K, R> templateSpec;
KafkaInboundGatewayListenerContainerSpec(KafkaMessageListenerContainerSpec<K, V> containerSpec,
KafkaTemplateSpec<K, R> templateSpec) {
super(containerSpec.get(), templateSpec.getTemplate());
this.containerSpec = containerSpec;
this.templateSpec = templateSpec;
}
/**
* Configure a listener container by invoking the {@link Consumer} callback, with a
* {@link KafkaMessageListenerContainerSpec} argument.
* @param configurer the configurer Java 8 Lambda.
* @return the spec.
*/
public KafkaInboundGatewayListenerContainerSpec<K, V, R> configureListenerContainer(
Consumer<KafkaMessageListenerContainerSpec<K, V>> configurer) {
Assert.notNull(configurer, "The 'configurer' cannot be null");
configurer.accept(this.containerSpec);
return _this();
}
/**
* Configure a template by invoking the {@link Consumer} callback, with a
* {@link KafkaTemplateSpec} argument.
* @param configurer the configurer Java 8 Lambda.
* @return the spec.
*/
public KafkaInboundGatewayListenerContainerSpec<K, V, R> configureTemplate(
Consumer<KafkaTemplateSpec<K, R>> configurer) {
Assert.notNull(configurer, "The 'configurer' cannot be null");
configurer.accept(this.templateSpec);
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return new ObjectStringMapBuilder()
.put(this.containerSpec.get(), this.containerSpec.getId())
.put(this.templateSpec.get(), this.templateSpec.getId())
.get();
}
}
}

233
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java
View File

@@ -19,24 +19,13 @@ package org.springframework.integration.kafka.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.integration.dsl.MessageProducerSpec;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.MessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
@@ -177,7 +166,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
KafkaMessageDrivenChannelAdapterListenerContainerSpec(KafkaMessageListenerContainerSpec<K, V> spec,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
super(spec.container, listenerMode);
super(spec.get(), listenerMode);
this.spec = spec;
}
@@ -196,225 +185,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.spec.container, this.spec.getId());
}
}
/**
* A helper class in the Builder pattern style to delegate options to the
* {@link ConcurrentMessageListenerContainer}.
*
* @param <K> the key type.
* @param <V> the value type.
*/
public static class KafkaMessageListenerContainerSpec<K, V>
extends IntegrationComponentSpec<KafkaMessageListenerContainerSpec<K, V>, ConcurrentMessageListenerContainer<K, V>> {
private final ConcurrentMessageListenerContainer<K, V> container;
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
ContainerProperties containerProperties) {
this.container = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
TopicPartitionInitialOffset... topicPartitions) {
this(consumerFactory, new ContainerProperties(topicPartitions));
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, String... topics) {
this(consumerFactory, new ContainerProperties(topics));
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
this(consumerFactory, new ContainerProperties(topicPattern));
}
@Override
public KafkaMessageListenerContainerSpec<K, V> id(String id) {
return super.id(id);
}
/**
* Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}.
* @param concurrency the concurrency maximum number.
* @return the spec.
* @see ConcurrentMessageListenerContainer#setConcurrency(int)
*/
public KafkaMessageListenerContainerSpec<K, V> concurrency(int concurrency) {
this.container.setConcurrency(concurrency);
return this;
}
/**
* Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}.
* @param errorHandler the {@link ErrorHandler}.
* @return the spec.
* @see ErrorHandler
*/
public KafkaMessageListenerContainerSpec<K, V> errorHandler(ErrorHandler errorHandler) {
this.container.getContainerProperties().setErrorHandler(errorHandler);
return this;
}
/**
* Set the ack mode to use when auto ack (in the configuration properties) is false.
* <ul>
* <li>RECORD: Ack after each record has been passed to the listener.</li>
* <li>BATCH: Ack after each batch of records received from the consumer has been
* passed to the listener</li>
* <li>TIME: Ack after this number of milliseconds; (should be greater than
* {@code #setPollTimeout(long) pollTimeout}.</li>
* <li>COUNT: Ack after at least this number of records have been received</li>
* <li>MANUAL: Listener is responsible for acking - use a
* {@link AcknowledgingMessageListener}.
* </ul>
* @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH.
* @return the spec.
* @see AbstractMessageListenerContainer.AckMode
*/
public KafkaMessageListenerContainerSpec<K, V> ackMode(AbstractMessageListenerContainer.AckMode ackMode) {
this.container.getContainerProperties().setAckMode(ackMode);
return this;
}
/**
* Set the max time to block in the consumer waiting for records.
* @param pollTimeout the timeout in ms; default 1000.
* @return the spec.
* @see ContainerProperties#setPollTimeout(long)
*/
public KafkaMessageListenerContainerSpec<K, V> pollTimeout(long pollTimeout) {
this.container.getContainerProperties().setPollTimeout(pollTimeout);
return this;
}
/**
* Set the number of outstanding record count after which offsets should be
* committed when {@link AbstractMessageListenerContainer.AckMode#COUNT}
* or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
* @param count the count
* @return the spec.
* @see ContainerProperties#setAckCount(int)
*/
public KafkaMessageListenerContainerSpec<K, V> ackCount(int count) {
this.container.getContainerProperties().setAckCount(count);
return this;
}
/**
* Set the time (ms) after which outstanding offsets should be committed when
* {@link AbstractMessageListenerContainer.AckMode#TIME} or
* {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
* Should be larger than zero.
* @param millis the time
* @return the spec.
* @see ContainerProperties#setAckTime(long)
*/
public KafkaMessageListenerContainerSpec<K, V> ackTime(long millis) {
this.container.getContainerProperties().setAckTime(millis);
return this;
}
/**
* Set the executor for threads that poll the consumer.
* @param consumerTaskExecutor the executor
* @return the spec.
* @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor)
*/
public KafkaMessageListenerContainerSpec<K, V> consumerTaskExecutor(
AsyncListenableTaskExecutor consumerTaskExecutor) {
this.container.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor);
return this;
}
/**
* Set the timeout for shutting down the container. This is the maximum amount of
* time that the invocation to {@code #stop(Runnable)} will block for, before
* returning.
* @param shutdownTimeout the shutdown timeout.
* @return the spec.
* @see ContainerProperties#setShutdownTimeout(long)
*/
public KafkaMessageListenerContainerSpec<K, V> shutdownTimeout(long shutdownTimeout) {
this.container.getContainerProperties().setShutdownTimeout(shutdownTimeout);
return this;
}
/**
* Set the user defined {@link ConsumerRebalanceListener} implementation.
* @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance
* @return the spec.
* @see ContainerProperties#setConsumerRebalanceListener(ConsumerRebalanceListener)
*/
public KafkaMessageListenerContainerSpec<K, V> consumerRebalanceListener(
ConsumerRebalanceListener consumerRebalanceListener) {
this.container.getContainerProperties().setConsumerRebalanceListener(consumerRebalanceListener);
return this;
}
/**
* Set the commit callback; by default a simple logging callback is used to log
* success at DEBUG level and failures at ERROR level.
* @param commitCallback the callback.
* @return the spec.
* @see ContainerProperties#setCommitCallback(OffsetCommitCallback)
*/
public KafkaMessageListenerContainerSpec<K, V> commitCallback(OffsetCommitCallback commitCallback) {
this.container.getContainerProperties().setCommitCallback(commitCallback);
return this;
}
/**
* Set whether or not to call consumer.commitSync() or commitAsync() when the
* container is responsible for commits. Default true. See
* https://github.com/spring-projects/spring-kafka/issues/62 At the time of
* writing, async commits are not entirely reliable.
* @param syncCommits true to use commitSync().
* @return the spec.
* @see ContainerProperties#setSyncCommits(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> syncCommits(boolean syncCommits) {
this.container.getContainerProperties().setSyncCommits(syncCommits);
return this;
}
/**
* Set the idle event interval; when set, an event is emitted if a poll returns
* no records and this interval has elapsed since a record was returned.
* @param idleEventInterval the interval.
* @return the spec.
* @see ContainerProperties#setIdleEventInterval(Long)
*/
public KafkaMessageListenerContainerSpec<K, V> idleEventInterval(Long idleEventInterval) {
this.container.getContainerProperties().setIdleEventInterval(idleEventInterval);
return this;
}
/**
* Set whether the container should ack messages that throw exceptions or not.
* @param ackOnError whether the container should acknowledge messages that throw
* exceptions.
* @return the spec.
* @see ContainerProperties#setAckOnError(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> ackOnError(boolean ackOnError) {
this.container.getContainerProperties().setAckOnError(ackOnError);
return this;
}
/**
* Set the group id for this container. Overrides any {@code group.id} property
* provided by the consumer factory configuration.
* @param groupId the group id.
* @return the spec.
* @see ContainerProperties#setAckOnError(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> groupId(String groupId) {
this.container.getContainerProperties().setGroupId(groupId);
return this;
return Collections.singletonMap(this.spec.get(), this.spec.getId());
}
}

256
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageListenerContainerSpec.java Normal file
View File

@@ -0,0 +1,256 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.dsl;
import java.util.regex.Pattern;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
/**
* A helper class in the Builder pattern style to delegate options to the
* {@link ConcurrentMessageListenerContainer}.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
*/
public class KafkaMessageListenerContainerSpec<K, V>
extends IntegrationComponentSpec<KafkaMessageListenerContainerSpec<K, V>, ConcurrentMessageListenerContainer<K, V>> {
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
ContainerProperties containerProperties) {
this.target = new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory,
TopicPartitionInitialOffset... topicPartitions) {
this(consumerFactory, new ContainerProperties(topicPartitions));
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, String... topics) {
this(consumerFactory, new ContainerProperties(topics));
}
KafkaMessageListenerContainerSpec(ConsumerFactory<K, V> consumerFactory, Pattern topicPattern) {
this(consumerFactory, new ContainerProperties(topicPattern));
}
@Override
public KafkaMessageListenerContainerSpec<K, V> id(String id) {
return super.id(id);
}
/**
* Specify a concurrency maximum number for the {@link AbstractMessageListenerContainer}.
* @param concurrency the concurrency maximum number.
* @return the spec.
* @see ConcurrentMessageListenerContainer#setConcurrency(int)
*/
public KafkaMessageListenerContainerSpec<K, V> concurrency(int concurrency) {
this.target.setConcurrency(concurrency);
return this;
}
/**
* Specify an {@link ErrorHandler} for the {@link AbstractMessageListenerContainer}.
* @param errorHandler the {@link ErrorHandler}.
* @return the spec.
* @see ErrorHandler
*/
public KafkaMessageListenerContainerSpec<K, V> errorHandler(ErrorHandler errorHandler) {
this.target.getContainerProperties().setErrorHandler(errorHandler);
return this;
}
/**
* Set the ack mode to use when auto ack (in the configuration properties) is false.
* <ul>
* <li>RECORD: Ack after each record has been passed to the listener.</li>
* <li>BATCH: Ack after each batch of records received from the consumer has been
* passed to the listener</li>
* <li>TIME: Ack after this number of milliseconds; (should be greater than
* {@code #setPollTimeout(long) pollTimeout}.</li>
* <li>COUNT: Ack after at least this number of records have been received</li>
* <li>MANUAL: Listener is responsible for acking - use a
* {@link AcknowledgingMessageListener}.
* </ul>
* @param ackMode the {@link AbstractMessageListenerContainer.AckMode}; default BATCH.
* @return the spec.
* @see AbstractMessageListenerContainer.AckMode
*/
public KafkaMessageListenerContainerSpec<K, V> ackMode(AbstractMessageListenerContainer.AckMode ackMode) {
this.target.getContainerProperties().setAckMode(ackMode);
return this;
}
/**
* Set the max time to block in the consumer waiting for records.
* @param pollTimeout the timeout in ms; default 1000.
* @return the spec.
* @see ContainerProperties#setPollTimeout(long)
*/
public KafkaMessageListenerContainerSpec<K, V> pollTimeout(long pollTimeout) {
this.target.getContainerProperties().setPollTimeout(pollTimeout);
return this;
}
/**
* Set the number of outstanding record count after which offsets should be
* committed when {@link AbstractMessageListenerContainer.AckMode#COUNT}
* or {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
* @param count the count
* @return the spec.
* @see ContainerProperties#setAckCount(int)
*/
public KafkaMessageListenerContainerSpec<K, V> ackCount(int count) {
this.target.getContainerProperties().setAckCount(count);
return this;
}
/**
* Set the time (ms) after which outstanding offsets should be committed when
* {@link AbstractMessageListenerContainer.AckMode#TIME} or
* {@link AbstractMessageListenerContainer.AckMode#COUNT_TIME} is being used.
* Should be larger than zero.
* @param millis the time
* @return the spec.
* @see ContainerProperties#setAckTime(long)
*/
public KafkaMessageListenerContainerSpec<K, V> ackTime(long millis) {
this.target.getContainerProperties().setAckTime(millis);
return this;
}
/**
* Set the executor for threads that poll the consumer.
* @param consumerTaskExecutor the executor
* @return the spec.
* @see ContainerProperties#setConsumerTaskExecutor(AsyncListenableTaskExecutor)
*/
public KafkaMessageListenerContainerSpec<K, V> consumerTaskExecutor(
AsyncListenableTaskExecutor consumerTaskExecutor) {
this.target.getContainerProperties().setConsumerTaskExecutor(consumerTaskExecutor);
return this;
}
/**
* Set the timeout for shutting down the container. This is the maximum amount of
* time that the invocation to {@code #stop(Runnable)} will block for, before
* returning.
* @param shutdownTimeout the shutdown timeout.
* @return the spec.
* @see ContainerProperties#setShutdownTimeout(long)
*/
public KafkaMessageListenerContainerSpec<K, V> shutdownTimeout(long shutdownTimeout) {
this.target.getContainerProperties().setShutdownTimeout(shutdownTimeout);
return this;
}
/**
* Set the user defined {@link ConsumerRebalanceListener} implementation.
* @param consumerRebalanceListener the {@link ConsumerRebalanceListener} instance
* @return the spec.
* @see ContainerProperties#setConsumerRebalanceListener(ConsumerRebalanceListener)
*/
public KafkaMessageListenerContainerSpec<K, V> consumerRebalanceListener(
ConsumerRebalanceListener consumerRebalanceListener) {
this.target.getContainerProperties().setConsumerRebalanceListener(consumerRebalanceListener);
return this;
}
/**
* Set the commit callback; by default a simple logging callback is used to log
* success at DEBUG level and failures at ERROR level.
* @param commitCallback the callback.
* @return the spec.
* @see ContainerProperties#setCommitCallback(OffsetCommitCallback)
*/
public KafkaMessageListenerContainerSpec<K, V> commitCallback(OffsetCommitCallback commitCallback) {
this.target.getContainerProperties().setCommitCallback(commitCallback);
return this;
}
/**
* Set whether or not to call consumer.commitSync() or commitAsync() when the
* container is responsible for commits. Default true. See
* https://github.com/spring-projects/spring-kafka/issues/62 At the time of
* writing, async commits are not entirely reliable.
* @param syncCommits true to use commitSync().
* @return the spec.
* @see ContainerProperties#setSyncCommits(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> syncCommits(boolean syncCommits) {
this.target.getContainerProperties().setSyncCommits(syncCommits);
return this;
}
/**
* Set the idle event interval; when set, an event is emitted if a poll returns
* no records and this interval has elapsed since a record was returned.
* @param idleEventInterval the interval.
* @return the spec.
* @see ContainerProperties#setIdleEventInterval(Long)
*/
public KafkaMessageListenerContainerSpec<K, V> idleEventInterval(Long idleEventInterval) {
this.target.getContainerProperties().setIdleEventInterval(idleEventInterval);
return this;
}
/**
* Set whether the container should ack messages that throw exceptions or not.
* @param ackOnError whether the container should acknowledge messages that throw
* exceptions.
* @return the spec.
* @see ContainerProperties#setAckOnError(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> ackOnError(boolean ackOnError) {
this.target.getContainerProperties().setAckOnError(ackOnError);
return this;
}
/**
* Set the group id for this container. Overrides any {@code group.id} property
* provided by the consumer factory configuration.
* @param groupId the group id.
* @return the spec.
* @see ContainerProperties#setAckOnError(boolean)
*/
public KafkaMessageListenerContainerSpec<K, V> groupId(String groupId) {
this.target.getContainerProperties().setGroupId(groupId);
return this;
}
}

135
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java Normal file
View File

@@ -0,0 +1,135 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.integration.dsl.MessageHandlerSpec;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.GenericMessageListenerContainer;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
/**
* A {@link MessageHandlerSpec} implementation for the {@link KafkaProducerMessageHandler}
* as a gateway.
* @param <K> the key type.
* @param <V> the outbound value type.
* @param <R> the reply value type.
* @param <S> the {@link KafkaProducerMessageHandlerSpec} extension type.
*
* @author Gary Russell
*
* @since 3.0.2
*
*/
public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpec<K, V, R, S>>
extends KafkaProducerMessageHandlerSpec<K, V, S> {
KafkaOutboundGatewaySpec(ReplyingKafkaTemplate<K, V, R> kafkaTemplate) {
super(kafkaTemplate);
}
/**
* Set a message converter for replies (when a gateway).
* @param messageConverter the converter.
* @return the spec.
*/
public S replyMessageConverter(RecordMessageConverter messageConverter) {
this.target.setReplyMessageConverter(messageConverter);
return _this();
}
/**
* A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension.
*
* @param <K> the key type.
* @param <V> the outbound value type.
* @param <R> the reply value type.
*/
public static class KafkaGatewayMessageHandlerTemplateSpec<K, V, R>
extends KafkaOutboundGatewaySpec<K, V, R, KafkaGatewayMessageHandlerTemplateSpec<K, V, R>>
implements ComponentsRegistration {
private final ReplyingKafkaTemplateSpec<K, V, R> kafkaTemplateSpec;
@SuppressWarnings("unchecked")
KafkaGatewayMessageHandlerTemplateSpec(ProducerFactory<K, V> producerFactory,
GenericMessageListenerContainer<K, R> replyContainer) {
super(new ReplyingKafkaTemplate<>(producerFactory, replyContainer));
this.kafkaTemplateSpec =
new ReplyingKafkaTemplateSpec<>((ReplyingKafkaTemplate<K, V, R>) this.target.getKafkaTemplate());
}
/**
* Configure a Kafka Template by invoking the {@link Consumer} callback, with a
* {@link KafkaTemplateSpec} argument.
* @param configurer the configurer Java 8 Lambda.
* @return the spec.
*/
public KafkaGatewayMessageHandlerTemplateSpec<K, V, R> configureKafkaTemplate(
Consumer<ReplyingKafkaTemplateSpec<K, V, R>> configurer) {
Assert.notNull(configurer, "The 'configurer' cannot be null");
configurer.accept(this.kafkaTemplateSpec);
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId());
}
}
/**
* An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}.
*
* @param <K> the key type.
* @param <V> the request value type.
* @param <R> the reply value type.
*/
public static class ReplyingKafkaTemplateSpec<K, V, R> extends KafkaTemplateSpec<K, V> {
ReplyingKafkaTemplateSpec(ReplyingKafkaTemplate<K, V, R> kafkaTemplate) {
super(kafkaTemplate);
}
@SuppressWarnings("unchecked")
ReplyingKafkaTemplateSpec<K, V, R> taskScheduler(TaskScheduler scheduler) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setTaskScheduler(scheduler);
return this;
}
@SuppressWarnings("unchecked")
ReplyingKafkaTemplateSpec<K, V, R> replyTimeout(long replyTimeout) {
((ReplyingKafkaTemplate<K, V, R>) this.target).setReplyTimeout(replyTimeout);
return this;
}
}
}

108
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -24,7 +24,6 @@ import java.util.function.Function;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.integration.dsl.MessageHandlerSpec;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.expression.ValueExpression;
@@ -32,10 +31,8 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.KafkaHeaderMapper;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
@@ -47,6 +44,7 @@ import org.springframework.util.Assert;
*
* @author Artem Bilan
* @author Biju Kunjummen
* @author Gary Russell
*
* @since 3.0
*/
@@ -268,6 +266,50 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
return _this();
}
/**
* Set the channel to which successful send results are sent.
* @param sendSuccessChannel the channel.
* @return the spec.
* @since 3.0.2
*/
public S sendSuccessChannel(MessageChannel sendSuccessChannel) {
this.target.setSendSuccessChannel(sendSuccessChannel);
return _this();
}
/**
* Set the channel to which successful send results are sent.
* @param sendSuccessChannel the channel name.
* @return the spec.
* @since 3.0.2
*/
public S sendSuccessChannel(String sendSuccessChannel) {
this.target.setSendSuccessChannelName(sendSuccessChannel);
return _this();
}
/**
* Set the channel to which failed send results are sent.
* @param sendFailureChannel the channel.
* @return the spec.
* @since 3.0.2
*/
public S sendFailureChannel(MessageChannel sendFailureChannel) {
this.target.setSendFailureChannel(sendFailureChannel);
return _this();
}
/**
* Set the channel to which failed send results are sent.
* @param sendFailureChannel the channel name.
* @return the spec.
* @since 3.0.2
*/
public S sendFailureChannel(String sendFailureChannel) {
this.target.setSendFailureChannelName(sendFailureChannel);
return _this();
}
/**
* A {@link KafkaTemplate}-based {@link KafkaProducerMessageHandlerSpec} extension.
*
@@ -287,7 +329,7 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
/**
* Configure a Kafka Template by invoking the {@link Consumer} callback, with a
* {@link KafkaProducerMessageHandlerSpec.KafkaTemplateSpec} argument.
* {@link KafkaTemplateSpec} argument.
* @param configurer the configurer Java 8 Lambda.
* @return the spec.
*/
@@ -305,59 +347,5 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
}
/**
* An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}.
*
* @param <K> the key type.
* @param <V> the value type.
*/
public static class KafkaTemplateSpec<K, V>
extends IntegrationComponentSpec<KafkaTemplateSpec<K, V>, KafkaTemplate<K, V>> {
KafkaTemplateSpec(KafkaTemplate<K, V> kafkaTemplate) {
this.target = kafkaTemplate;
}
@Override
public KafkaTemplateSpec<K, V> id(String id) {
return super.id(id);
}
/**
/**
* Set the default topic for send methods where a topic is not
* providing.
* @param defaultTopic the topic.
* @return the spec
*/
public KafkaTemplateSpec<K, V> defaultTopic(String defaultTopic) {
this.target.setDefaultTopic(defaultTopic);
return this;
}
/**
* Set a {@link ProducerListener} which will be invoked when Kafka acknowledges
* a send operation. By default a {@link LoggingProducerListener} is configured
* which logs errors only.
* @param producerListener the listener; may be {@code null}.
* @return the spec
*/
public KafkaTemplateSpec<K, V> producerListener(ProducerListener<K, V> producerListener) {
this.target.setProducerListener(producerListener);
return this;
}
/**
* Set the message converter to use.
* @param messageConverter the message converter.
* @return the spec
*/
public KafkaTemplateSpec<K, V> messageConverter(RecordMessageConverter messageConverter) {
this.target.setMessageConverter(messageConverter);
return this;
}
}
}

91
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaTemplateSpec.java Normal file
View File

@@ -0,0 +1,91 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.dsl;
import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.converter.RecordMessageConverter;
/**
* An {@link IntegrationComponentSpec} implementation for the {@link KafkaTemplate}.
*
* @param <K> the key type.
* @param <V> the value type.
*
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
*/
public class KafkaTemplateSpec<K, V>
extends IntegrationComponentSpec<KafkaTemplateSpec<K, V>, KafkaTemplate<K, V>> {
KafkaTemplateSpec(KafkaTemplate<K, V> kafkaTemplate) {
this.target = kafkaTemplate;
}
KafkaTemplateSpec(ProducerFactory<K, V> producerFactory) {
this.target = new KafkaTemplate<>(producerFactory);
}
public KafkaTemplate<K, V> getTemplate() {
return this.target;
}
@Override
public KafkaTemplateSpec<K, V> id(String id) {
return super.id(id);
}
/**
/**
* Set the default topic for send methods where a topic is not
* providing.
* @param defaultTopic the topic.
* @return the spec
*/
public KafkaTemplateSpec<K, V> defaultTopic(String defaultTopic) {
this.target.setDefaultTopic(defaultTopic);
return this;
}
/**
* Set a {@link ProducerListener} which will be invoked when Kafka acknowledges
* a send operation. By default a {@link LoggingProducerListener} is configured
* which logs errors only.
* @param producerListener the listener; may be {@code null}.
* @return the spec
*/
public KafkaTemplateSpec<K, V> producerListener(ProducerListener<K, V> producerListener) {
this.target.setProducerListener(producerListener);
return this;
}
/**
* Set the message converter to use.
* @param messageConverter the message converter.
* @return the spec
*/
public KafkaTemplateSpec<K, V> messageConverter(RecordMessageConverter messageConverter) {
this.target.setMessageConverter(messageConverter);
return this;
}
}

318
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java Normal file
View File

@@ -0,0 +1,318 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.inbound;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.Assert;
/**
* Inbound gateway.
*
* @param <K> the key type.
* @param <V> the request value type.
* @param <R> the reply value type.
*
* @author Gary Russell
*
* @since 3.0.2
*
*/
public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implements OrderlyShutdownCapable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
private final IntegrationRecordMessageListener listener = new IntegrationRecordMessageListener();
private final AbstractMessageListenerContainer<K, V> messageListenerContainer;
private final KafkaTemplate<K, R> kafkaTemplate;
private RetryTemplate retryTemplate;
private RecoveryCallback<? extends Object> recoveryCallback;
/**
* Construct an instance with the provided container.
* @param messageListenerContainer the container.
* @param kafkaTemplate the kafka template.
*/
public KafkaInboundGateway(AbstractMessageListenerContainer<K, V> messageListenerContainer,
KafkaTemplate<K, R> kafkaTemplate) {
Assert.notNull(messageListenerContainer, "messageListenerContainer is required");
Assert.notNull(kafkaTemplate, "kafkaTemplate is required");
Assert.isNull(messageListenerContainer.getContainerProperties().getMessageListener(),
"Container must not already have a listener");
this.messageListenerContainer = messageListenerContainer;
this.messageListenerContainer.setAutoStartup(false);
this.kafkaTemplate = kafkaTemplate;
setErrorMessageStrategy(new RawRecordHeaderErrorMessageStrategy());
}
/**
* Set the message converter; must be a {@link RecordMessageConverter} or
* {@link BatchMessageConverter} depending on mode.
* @param messageConverter the converter.
*/
public void setMessageConverter(RecordMessageConverter messageConverter) {
this.listener.setMessageConverter(messageConverter);
}
/**
* When using a type-aware message converter (such as {@code StringJsonMessageConverter},
* set the payload type the converter should create. Defaults to {@link Object}.
* @param payloadType the type.
*/
public void setPayloadType(Class<?> payloadType) {
this.listener.setFallbackType(payloadType);
}
/**
* Specify a {@link RetryTemplate} instance to wrap
* {@link KafkaInboundGateway.IntegrationRecordMessageListener} into
* {@link RetryingMessageListenerAdapter}.
* @param retryTemplate the {@link RetryTemplate} to use.
*/
public void setRetryTemplate(RetryTemplate retryTemplate) {
this.retryTemplate = retryTemplate;
}
/**
* A {@link RecoveryCallback} instance for retry operation;
* if null, the exception will be thrown to the container after retries are exhausted
* (unless an error channel is configured).
* Does not make sense if {@link #setRetryTemplate(RetryTemplate)} isn't specified.
* @param recoveryCallback the recovery callback.
*/
public void setRecoveryCallback(RecoveryCallback<? extends Object> recoveryCallback) {
this.recoveryCallback = recoveryCallback;
}
@Override
protected void onInit() throws Exception {
super.onInit();
MessageListener<K, V> listener = this.listener;
if (this.retryTemplate != null) {
listener = new RetryingMessageListenerAdapter<>(listener, this.retryTemplate,
this.recoveryCallback);
this.retryTemplate.registerListener(this.listener);
}
this.messageListenerContainer.getContainerProperties().setMessageListener(listener);
}
@Override
protected void doStart() {
this.messageListenerContainer.start();
}
@Override
protected void doStop() {
this.messageListenerContainer.stop();
}
@Override
public String getComponentType() {
return "kafka:inbound-gateway";
}
@Override
public int beforeShutdown() {
this.messageListenerContainer.stop();
return getPhase();
}
@Override
public int afterShutdown() {
return getPhase();
}
/**
* If there's a retry template, it will set the attributes holder via the listener. If
* there's no retry template, but there's an error channel, we create a new attributes
* holder here. If an attributes holder exists (by either method), we set the
* attributes for use by the {@link ErrorMessageStrategy}.
* @param record the record.
* @param message the message.
*/
private void setAttributesIfNecessary(Object record, Message<?> message) {
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder | this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}
if (needAttributes) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes != null) {
attributes.setAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY, message);
attributes.setAttribute(KafkaHeaders.RAW_DATA, record);
}
}
}
@Override
protected AttributeAccessor getErrorMessageAttributes(Message<?> message) {
AttributeAccessor attributes = attributesHolder.get();
if (attributes == null) {
return super.getErrorMessageAttributes(message);
}
else {
return attributes;
}
}
private class IntegrationRecordMessageListener extends RecordMessagingMessageListenerAdapter<K, V>
implements RetryListener {
IntegrationRecordMessageListener() {
super(null, null);
}
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = null;
try {
message = toMessagingMessage(record, acknowledgment, consumer);
if (KafkaInboundGateway.this.retryTemplate != null) {
message = addDeliveryAttemptHeader(message);
}
setAttributesIfNecessary(record, message);
}
catch (RuntimeException e) {
if (getErrorChannel() != null) {
KafkaInboundGateway.this.messagingTemplate.send(getErrorChannel(), buildErrorMessage(null,
new ConversionException("Failed to convert to message for: " + record, e)));
}
}
if (message != null) {
try {
Message<?> reply = sendAndReceiveMessage(message);
if (reply != null) {
reply = enhanceReply(message, reply);
KafkaInboundGateway.this.kafkaTemplate.send(reply);
}
}
finally {
if (KafkaInboundGateway.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaInboundGateway.this.logger.debug("Converter returned a null message for: "
+ record);
}
}
private Message<?> addDeliveryAttemptHeader(Message<?> message) {
Message<?> messageToReturn = message;
AtomicInteger deliveryAttempt =
new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
if (message.getHeaders() instanceof KafkaMessageHeaders) {
((KafkaMessageHeaders) message.getHeaders()).getRawHeaders()
.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
}
else {
messageToReturn = MessageBuilder.fromMessage(message)
.setHeader(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt)
.build();
}
return messageToReturn;
}
private Message<?> enhanceReply(Message<?> message, Message<?> reply) {
AbstractIntegrationMessageBuilder<?> builder = null;
MessageHeaders replyHeaders = reply.getHeaders();
MessageHeaders requestHeaders = message.getHeaders();
if (replyHeaders.get(KafkaHeaders.CORRELATION_ID) == null &&
requestHeaders.get(KafkaHeaders.CORRELATION_ID) != null) {
builder = getMessageBuilderFactory().fromMessage(reply)
.setHeader(KafkaHeaders.CORRELATION_ID, requestHeaders.get(KafkaHeaders.CORRELATION_ID));
}
if (replyHeaders.get(KafkaHeaders.TOPIC) == null &&
requestHeaders.get(KafkaHeaders.REPLY_TOPIC) != null) {
if (builder == null) {
builder = getMessageBuilderFactory().fromMessage(reply);
}
builder.setHeader(KafkaHeaders.TOPIC, requestHeaders.get(KafkaHeaders.REPLY_TOPIC));
}
if (replyHeaders.get(KafkaHeaders.PARTITION_ID) == null &&
requestHeaders.get(KafkaHeaders.REPLY_PARTITION) != null) {
if (builder == null) {
builder = getMessageBuilderFactory().fromMessage(reply);
}
builder.setHeader(KafkaHeaders.PARTITION_ID, requestHeaders.get(KafkaHeaders.REPLY_PARTITION));
}
if (builder != null) {
return builder.build();
}
return reply;
}
@Override
public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
if (KafkaInboundGateway.this.recoveryCallback != null) {
attributesHolder.set(context);
}
return true;
}
@Override
public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
attributesHolder.remove();
}
@Override
public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback,
Throwable throwable) {
// Empty
}
}
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -305,9 +305,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
* @since 2.1.1
*/
private void setAttributesIfNecessary(Object record, Message<?> message) {
boolean needHolder = getErrorChannel() != null
&& KafkaMessageDrivenChannelAdapter.this.retryTemplate == null;
boolean needAttributes = needHolder | KafkaMessageDrivenChannelAdapter.this.retryTemplate != null;
boolean needHolder = getErrorChannel() != null && this.retryTemplate == null;
boolean needAttributes = needHolder | this.retryTemplate != null;
if (needHolder) {
attributesHolder.set(ErrorMessageUtils.getAttributeAccessor(null, null));
}

295
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2017 the original author or authors.
* Copyright 2013-2018 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.
@@ -16,11 +16,23 @@
package org.springframework.integration.kafka.outbound;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.expression.EvaluationContext;
@@ -28,27 +40,38 @@ import org.springframework.expression.Expression;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.kafka.support.KafkaSendFailureException;
import org.springframework.integration.support.DefaultErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.JacksonPresent;
import org.springframework.kafka.support.KafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.support.SimpleKafkaHeaderMapper;
import org.springframework.kafka.support.converter.KafkaMessageHeaders;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* Kafka Message Handler.
* Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate}
* it is used as the handler in an outbound gateway. When supplied with a simple
* {@link KafkaTemplate} it used as the handler in an outbound channel adapter.
*
* @param <K> the key type.
* @param <V> the value type.
@@ -61,12 +84,16 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
*
* @since 0.5
*/
public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingHandler {
public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMessageHandler {
private static final long DEFAULT_SEND_TIMEOUT = 10000;
private final Map<String, Set<Integer>> replyTopicsAndPartitions = new HashMap<>();
private final KafkaTemplate<K, V> kafkaTemplate;
private final boolean isGateway;
private EvaluationContext evaluationContext;
private volatile Expression topicExpression;
@@ -83,18 +110,37 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
private KafkaHeaderMapper headerMapper;
private RecordMessageConverter replyMessageConverter = new MessagingMessageConverter();
private MessageChannel sendFailureChannel;
private String sendFailureChannelName;
private MessageChannel sendSuccessChannel;
private String sendSuccessChannelName;
private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy();
private Type replyPayloadType = Object.class;
private volatile boolean noOutputChannel;
public KafkaProducerMessageHandler(final KafkaTemplate<K, V> kafkaTemplate) {
Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null");
this.kafkaTemplate = kafkaTemplate;
this.isGateway = kafkaTemplate instanceof ReplyingKafkaTemplate;
if (this.isGateway) {
setAsync(true);
updateNotPropagatedHeaders(
new String[] { KafkaHeaders.TOPIC, KafkaHeaders.PARTITION_ID, KafkaHeaders.MESSAGE_KEY }, false);
}
if (JacksonPresent.isJackson2Present()) {
this.headerMapper = new DefaultKafkaHeaderMapper();
}
else {
this.headerMapper = new SimpleKafkaHeaderMapper();
}
}
public void setTopicExpression(Expression topicExpression) {
@@ -193,6 +239,24 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
this.sendFailureChannelName = sendFailureChannelName;
}
/**
* Set the success channel.
* @param sendSuccessChannel the Success channel.
* @since 3.0.2
*/
public void setSendSuccessChannel(MessageChannel sendSuccessChannel) {
this.sendSuccessChannel = sendSuccessChannel;
}
/**
* Set the Success channel name.
* @param sendSuccessChannelName the Success channel name.
* @since 3.0.2
*/
public void setSendSuccessChannelName(String sendSuccessChannelName) {
this.sendSuccessChannelName = sendSuccessChannelName;
}
/**
* Set the error message strategy implementation to use when sending error messages after
* send failures. Cannot be null.
@@ -204,6 +268,34 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
this.errorMessageStrategy = errorMessageStrategy;
}
/**
* Set a message converter for gateway replies.
* @param messageConverter the converter.
* @since 3.0.2
* @see #setReplyPayloadType(Type)
*/
public void setReplyMessageConverter(RecordMessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' cannot be null");
this.replyMessageConverter = messageConverter;
}
/**
* When using a type-aware message converter (such as {@code StringJsonMessageConverter},
* set the payload type the converter should create. Defaults to {@link Object}.
* @param payloadType the type.
* @since 3.0.2
* @see #setReplyMessageConverter(RecordMessageConverter)
*/
public void setReplyPayloadType(Type payloadType) {
Assert.notNull(payloadType, "'payloadType' cannot be null");
this.replyPayloadType = payloadType;
}
@Override
public String getComponentType() {
return this.isGateway ? "kafka:outbound-gateway" : "kafka:outbound-channel-adapter";
}
protected MessageChannel getSendFailureChannel() {
if (this.sendFailureChannel != null) {
return this.sendFailureChannel;
@@ -215,32 +307,43 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
return null;
}
protected MessageChannel getSendSuccessChannel() {
if (this.sendSuccessChannel != null) {
return this.sendSuccessChannel;
}
else if (this.sendSuccessChannelName != null) {
this.sendSuccessChannel = getChannelResolver().resolveDestination(this.sendSuccessChannelName);
return this.sendSuccessChannel;
}
return null;
}
@Override
protected void onInit() throws Exception {
super.onInit();
protected void doInit() {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@SuppressWarnings("unchecked")
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
protected Object handleRequestMessage(final Message<?> message) {
MessageHeaders messageHeaders = message.getHeaders();
String topic = this.topicExpression != null ?
this.topicExpression.getValue(this.evaluationContext, message, String.class)
: message.getHeaders().get(KafkaHeaders.TOPIC, String.class);
: messageHeaders.get(KafkaHeaders.TOPIC, String.class);
Assert.state(StringUtils.hasText(topic), "The 'topic' can not be empty or null");
Integer partitionId = this.partitionIdExpression != null ?
this.partitionIdExpression.getValue(this.evaluationContext, message, Integer.class)
: message.getHeaders().get(KafkaHeaders.PARTITION_ID, Integer.class);
: messageHeaders.get(KafkaHeaders.PARTITION_ID, Integer.class);
Object messageKey = this.messageKeyExpression != null
? this.messageKeyExpression.getValue(this.evaluationContext, message)
: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
: messageHeaders.get(KafkaHeaders.MESSAGE_KEY);
Long timestamp = this.timestampExpression != null
? this.timestampExpression.getValue(this.evaluationContext, message, Long.class)
: message.getHeaders().get(KafkaHeaders.TIMESTAMP, Long.class);
: messageHeaders.get(KafkaHeaders.TIMESTAMP, Long.class);
V payload = (V) message.getPayload();
if (payload instanceof KafkaNull) {
@@ -250,21 +353,120 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
Headers headers = null;
if (this.headerMapper != null) {
headers = new RecordHeaders();
this.headerMapper.fromHeaders(message.getHeaders(), headers);
this.headerMapper.fromHeaders(messageHeaders, headers);
}
final ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, partitionId, timestamp,
final ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partitionId, timestamp,
(K) messageKey, payload, headers);
ListenableFuture<SendResult<K, V>> future = this.kafkaTemplate.send(producerRecord);
if (getSendFailureChannel() != null || getOutputChannel() != null) {
ListenableFuture<SendResult<K, V>> sendFuture;
RequestReplyFuture<K, V, Object> gatewayFuture = null;
MessageChannel metadataChannel = null;
if (this.isGateway) {
metadataChannel = getSendSuccessChannel();
producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message)));
gatewayFuture = ((ReplyingKafkaTemplate<K, V, Object>) this.kafkaTemplate).sendAndReceive(producerRecord);
sendFuture = gatewayFuture.getSendFuture();
}
else {
sendFuture = this.kafkaTemplate.send(producerRecord);
// TODO: In 3.1, always use the success channel.
if (!this.noOutputChannel) {
metadataChannel = getOutputChannel();
if (metadataChannel == null) {
this.noOutputChannel = true;
}
}
if (metadataChannel == null) {
metadataChannel = getSendSuccessChannel();
}
}
try {
processSendResult(message, producerRecord, sendFuture, metadataChannel);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new MessageHandlingException(message, e);
}
catch (ExecutionException e) {
// TODO: in 3.1 change this to e.getCause()
throw new MessageHandlingException(message, e);
}
return processReplyFuture(gatewayFuture);
}
private byte[] getReplyTopic(final Message<?> message) {
if (this.replyTopicsAndPartitions.isEmpty()) {
determineValidReplyTopicsAndPartitions();
}
Object replyHeader = message.getHeaders().get(KafkaHeaders.REPLY_TOPIC);
byte[] replyTopic = null;
String topicToCheck = null;
if (replyHeader instanceof String) {
replyTopic = ((String) replyHeader).getBytes(StandardCharsets.UTF_8);
topicToCheck = (String) replyHeader;
}
else if (replyHeader instanceof byte[]) {
replyTopic = (byte[]) replyHeader;
}
else if (replyHeader != null) {
throw new IllegalStateException(KafkaHeaders.REPLY_TOPIC + " must be String or byte[]");
}
if (replyTopic == null) {
if (this.replyTopicsAndPartitions.size() == 1) {
replyTopic = this.replyTopicsAndPartitions.keySet().iterator().next().getBytes(StandardCharsets.UTF_8);
}
else {
throw new IllegalStateException("No reply topic header and no default reply topic is can be determined");
}
}
else {
if (topicToCheck == null) {
topicToCheck = new String(replyTopic, StandardCharsets.UTF_8);
}
if (!this.replyTopicsAndPartitions.keySet().contains(topicToCheck)) {
throw new IllegalStateException("The reply topic header ["
+ topicToCheck +
"] does not match any reply container topic: " + this.replyTopicsAndPartitions.keySet());
}
}
Integer replyPartition = message.getHeaders().get(KafkaHeaders.REPLY_PARTITION, Integer.class);
if (replyPartition != null) {
if (topicToCheck == null) {
topicToCheck = new String(replyTopic, StandardCharsets.UTF_8);
}
if (!this.replyTopicsAndPartitions.get(topicToCheck).contains(replyPartition)) {
throw new IllegalStateException("The reply partition header ["
+ replyPartition + "] does not match any reply container partition for topic ["
+ topicToCheck + "]: " + this.replyTopicsAndPartitions.get(topicToCheck));
}
}
return replyTopic;
}
private void determineValidReplyTopicsAndPartitions() {
ReplyingKafkaTemplate<?, ?, ?> rkt = (ReplyingKafkaTemplate<?, ?, ?>) kafkaTemplate;
Collection<TopicPartition> replyTopics = rkt.getAssignedReplyTopicPartitions();
Map<String, Set<Integer>> topicsAndPartitions = new HashMap<>();
if (replyTopics != null) {
replyTopics.forEach(tp -> {
topicsAndPartitions.computeIfAbsent(tp.topic(), (k) -> new TreeSet<>());
topicsAndPartitions.get(tp.topic()).add(tp.partition());
});
this.replyTopicsAndPartitions.putAll(topicsAndPartitions);
}
}
public void processSendResult(final Message<?> message, final ProducerRecord<K, V> producerRecord,
ListenableFuture<SendResult<K, V>> future, MessageChannel metadataChannel)
throws InterruptedException, ExecutionException {
if (getSendFailureChannel() != null || metadataChannel != null) {
future.addCallback(new ListenableFutureCallback<SendResult<K, V>>() {
@Override
public void onSuccess(SendResult<K, V> result) {
if (getOutputChannel() != null) {
KafkaProducerMessageHandler.this.messagingTemplate.send(getOutputChannel(),
if (metadataChannel != null) {
KafkaProducerMessageHandler.this.messagingTemplate.send(metadataChannel,
getMessageBuilderFactory().fromMessage(message)
// TODO: change to constant when available
.setHeader("kafka_recordMetadata", result.getRecordMetadata()).build());
.setHeader(KafkaHeaders.RECORD_METADATA, result.getRecordMetadata()).build());
}
}
@@ -296,9 +498,58 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractMessageProducingH
}
}
@Override
public String getComponentType() {
return "kafka:outbound-channel-adapter";
private Future<?> processReplyFuture(RequestReplyFuture<?, ?, Object> future) {
if (future == null) {
return null;
}
return new ConvertingReplyFuture(future);
}
private final class ConvertingReplyFuture extends SettableListenableFuture<Object> {
ConvertingReplyFuture(RequestReplyFuture<?, ?, Object> future) {
addCallback(future);
}
private void addCallback(final RequestReplyFuture<?, ?, Object> future) {
future.addCallback(new ListenableFutureCallback<ConsumerRecord<?, Object>>() {
@Override
public void onSuccess(ConsumerRecord<?, Object> result) {
try {
set(dontLeakHeaders(KafkaProducerMessageHandler.this.replyMessageConverter.toMessage(result,
null, null, KafkaProducerMessageHandler.this.replyPayloadType)));
}
catch (Exception e) {
setException(e);
}
}
private Message<?> dontLeakHeaders(Message<?> message) {
if (message.getHeaders() instanceof KafkaMessageHeaders) {
Map<String, Object> headers = ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders();
headers.remove(KafkaHeaders.CORRELATION_ID);
headers.remove(KafkaHeaders.REPLY_TOPIC);
headers.remove(KafkaHeaders.REPLY_PARTITION);
return message;
}
else {
return getMessageBuilderFactory().fromMessage(message)
.removeHeader(KafkaHeaders.CORRELATION_ID)
.removeHeader(KafkaHeaders.REPLY_TOPIC)
.removeHeader(KafkaHeaders.REPLY_PARTITION)
.build();
}
}
@Override
public void onFailure(Throwable ex) {
setException(ex);
}
});
}
}
}

80
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.kafka.dsl;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
@@ -26,6 +27,8 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -48,13 +51,17 @@ import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandle
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.GenericMessageListenerContainer;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
@@ -64,6 +71,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.retry.support.RetryTemplate;
@@ -88,8 +96,13 @@ public class KafkaDslTests {
private static final String TEST_TOPIC3 = "test-topic3";
private static final String TEST_TOPIC4 = "test-topic4";
private static final String TEST_TOPIC5 = "test-topic5";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3);
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3,
TEST_TOPIC4, TEST_TOPIC5);
@Autowired
@Qualifier("sendToKafkaFlow.input")
@@ -130,6 +143,9 @@ public class KafkaDslTests {
@Autowired
private ContextConfiguration config;
@Autowired
private Gate gate;
@Test
public void testKafkaAdapters() throws Exception {
@@ -191,15 +207,24 @@ public class KafkaDslTests {
assertThat(this.kafkaTemplateTopic2).isNotNull();
this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo");
assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.config.fromSource).isEqualTo("foo");
}
@Test
public void testGateways() throws Exception {
assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS));
assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO");
}
@Configuration
@EnableIntegration
@EnableKafka
public static class ContextConfiguration {
private final CountDownLatch latch = new CountDownLatch(1);
private final CountDownLatch sourceFlowLatch = new CountDownLatch(1);
private final CountDownLatch replyContainerLatch = new CountDownLatch(1);
private Object fromSource;
@@ -301,12 +326,59 @@ public class KafkaDslTests {
e -> e.poller(Pollers.fixedDelay(100)))
.handle(p -> {
this.fromSource = p.getPayload();
this.latch.countDown();
this.sourceFlowLatch.countDown();
})
.get();
}
@Bean
public IntegrationFlow outboundGateFlow() {
return IntegrationFlows.from(Gate.class)
.handle(Kafka.outboundGateway(producerFactory(), replyContainer())
.configureKafkaTemplate(t -> t.replyTimeout(30_000)))
.get();
}
private GenericMessageListenerContainer<Integer, String> replyContainer() {
ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC5);
containerProperties.setGroupId("outGate");
containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// empty
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
ContextConfiguration.this.replyContainerLatch.countDown();
}
});
return new KafkaMessageListenerContainer<>(consumerFactory(), containerProperties);
}
@Bean
public IntegrationFlow serverGateway() {
return IntegrationFlows
.from(Kafka.inboundGateway(consumerFactory(), containerProperties(),
producerFactory()))
.<String, String>transform(String::toUpperCase)
.get();
}
private ContainerProperties containerProperties() {
ContainerProperties containerProperties = new ContainerProperties(TEST_TOPIC4);
containerProperties.setGroupId("inGateGroup");
return containerProperties;
}
}
public interface Gate {
String exchange(@Header(KafkaHeaders.TOPIC) String topic, String out);
}
}

306
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java Normal file
View File

@@ -0,0 +1,306 @@
/*
* Copyright 2018 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
*
* http://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.integration.kafka.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
import static org.springframework.kafka.test.assertj.KafkaConditions.value;
import java.lang.reflect.Type;
import java.util.Map;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.junit.ClassRule;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.converter.MessagingMessageConverter;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.ContainerTestUtils;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.retry.backoff.NoBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
/**
*
* @author Gary Russell
*
* @since 3.0.2
*
*/
public class InboundGatewayTests {
private static String topic1 = "testTopic1";
private static String topic2 = "testTopic2";
private static String topic3 = "testTopic3";
private static String topic4 = "testTopic4";
private static String topic5 = "testTopic5";
private static String topic6 = "testTopic6";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
topic6);
@Test
public void testInbound() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic1);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic1);
KafkaInboundGateway<Integer, String, String> gateway = new KafkaInboundGateway<>(container, template);
QueueChannel out = new QueueChannel();
DirectChannel reply = new DirectChannel();
gateway.setRequestChannel(out);
gateway.setReplyChannel(reply);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setReplyTimeout(30_000);
gateway.afterPropertiesSet();
gateway.setMessageConverter(new MessagingMessageConverter() {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic2)
.setHeader(KafkaHeaders.REPLY_PARTITION, 1)
.build();
}
});
gateway.start();
ContainerTestUtils.waitForAssignment(container, 2);
template.sendDefault(0, 1487694048607L, 1, "foo");
Message<?> received = out.receive(30_000);
assertThat(received).isNotNull();
MessageHeaders headers = received.getHeaders();
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic1);
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L);
assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME");
assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic2);
assertThat(headers.get("testHeader")).isEqualTo("testValue");
reply.send(MessageBuilder.withPayload("FOO").copyHeaders(headers).build());
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic2);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic2);
assertThat(record).has(partition(1));
assertThat(record).has(value("FOO"));
gateway.stop();
}
@Test
public void testInboundErrorRecover() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic3);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic3);
KafkaInboundGateway<Integer, String, String> gateway = new KafkaInboundGateway<>(container, template);
MessageChannel out = new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
throw new RuntimeException("intended");
}
};
QueueChannel errors = new QueueChannel();
gateway.setRequestChannel(out);
gateway.setErrorChannel(errors);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setMessageConverter(new MessagingMessageConverter() {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic4)
.setHeader(KafkaHeaders.REPLY_PARTITION, 1)
.build();
}
});
gateway.setReplyTimeout(30_000);
gateway.afterPropertiesSet();
gateway.start();
ContainerTestUtils.waitForAssignment(container, 2);
template.sendDefault(0, 1487694048607L, 1, "foo");
ErrorMessage em = (ErrorMessage) errors.receive(30_000);
assertThat(em).isNotNull();
Message<?> failed = ((MessagingException) em.getPayload()).getFailedMessage();
assertThat(failed).isNotNull();
MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel();
MessageHeaders headers = failed.getHeaders();
reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build());
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic3);
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L);
assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME");
assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic4);
assertThat(headers.get("testHeader")).isEqualTo("testValue");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic4);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic4);
assertThat(record).has(partition(1));
assertThat(record).has(value("ERROR"));
gateway.stop();
}
@Test
public void testInboundRetryErrorRecover() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic5);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic5);
KafkaInboundGateway<Integer, String, String> gateway = new KafkaInboundGateway<>(container, template);
MessageChannel out = new DirectChannel() {
@Override
protected boolean doSend(Message<?> message, long timeout) {
throw new RuntimeException("intended");
}
};
QueueChannel errors = new QueueChannel();
gateway.setRequestChannel(out);
gateway.setErrorChannel(errors);
gateway.setBeanFactory(mock(BeanFactory.class));
gateway.setMessageConverter(new MessagingMessageConverter() {
@Override
public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment,
Consumer<?, ?> consumer, Type type) {
Message<?> message = super.toMessage(record, acknowledgment, consumer, type);
return MessageBuilder.fromMessage(message)
.setHeader("testHeader", "testValue")
.setHeader(KafkaHeaders.REPLY_TOPIC, topic6)
.setHeader(KafkaHeaders.REPLY_PARTITION, 1)
.build();
}
});
gateway.setReplyTimeout(30_000);
RetryTemplate retryTemplate = new RetryTemplate();
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(2);
retryTemplate.setRetryPolicy(retryPolicy);
retryTemplate.setBackOffPolicy(new NoBackOffPolicy());
gateway.setRetryTemplate(retryTemplate);
gateway.setRecoveryCallback(
new ErrorMessageSendingRecoverer(errors, new RawRecordHeaderErrorMessageStrategy()));
gateway.afterPropertiesSet();
gateway.start();
ContainerTestUtils.waitForAssignment(container, 2);
template.sendDefault(0, 1487694048607L, 1, "foo");
ErrorMessage em = (ErrorMessage) errors.receive(30_000);
assertThat(em).isNotNull();
Message<?> failed = ((MessagingException) em.getPayload()).getFailedMessage();
assertThat(failed).isNotNull();
MessageChannel reply = (MessageChannel) em.getHeaders().getReplyChannel();
MessageHeaders headers = failed.getHeaders();
reply.send(MessageBuilder.withPayload("ERROR").copyHeaders(headers).build());
assertThat(headers.get(KafkaHeaders.RECEIVED_MESSAGE_KEY)).isEqualTo(1);
assertThat(headers.get(KafkaHeaders.RECEIVED_TOPIC)).isEqualTo(topic5);
assertThat(headers.get(KafkaHeaders.RECEIVED_PARTITION_ID)).isEqualTo(0);
assertThat(headers.get(KafkaHeaders.OFFSET)).isEqualTo(0L);
assertThat(headers.get(KafkaHeaders.RECEIVED_TIMESTAMP)).isEqualTo(1487694048607L);
assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME");
assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic6);
assertThat(headers.get("testHeader")).isEqualTo("testValue");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
Consumer<Integer, String> consumer = cf2.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic6);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic6);
assertThat(record).has(partition(1));
assertThat(record).has(value("ERROR"));
gateway.stop();
}
}

137
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -17,19 +17,26 @@
package org.springframework.integration.kafka.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.mockito.Mockito.mock;
import static org.springframework.kafka.test.assertj.KafkaConditions.key;
import static org.springframework.kafka.test.assertj.KafkaConditions.partition;
import static org.springframework.kafka.test.assertj.KafkaConditions.timestamp;
import static org.springframework.kafka.test.assertj.KafkaConditions.value;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
@@ -46,6 +53,9 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
@@ -53,6 +63,7 @@ import org.springframework.kafka.support.SendResult;
import org.springframework.kafka.test.rule.KafkaEmbedded;
import org.springframework.kafka.test.utils.KafkaTestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
@@ -71,9 +82,17 @@ public class KafkaProducerMessageHandlerTests {
private static String topic2 = "testTopic2out";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2);
private static String topic3 = "testTopic3out";
private static String topic4 = "testTopic4out";
private static String topic5 = "testTopic5out";
private static String topic6 = "testTopic6in";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
topic6);
private static Consumer<Integer, String> consumer;
@@ -85,6 +104,11 @@ public class KafkaProducerMessageHandlerTests {
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
}
@AfterClass
public static void tearDown() {
consumer.close();
}
@Test
public void testOutbound() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
@@ -149,7 +173,7 @@ public class KafkaProducerMessageHandlerTests {
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic1)
.setHeader(KafkaHeaders.TOPIC, topic2)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.setHeader(KafkaHeaders.TIMESTAMP, 1487694048607L)
@@ -157,7 +181,7 @@ public class KafkaProducerMessageHandlerTests {
.build();
handler.handleMessage(message);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic1);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic2);
assertThat(record).has(key(2));
assertThat(record).has(partition(1));
assertThat(record).has(value("foo"));
@@ -178,7 +202,7 @@ public class KafkaProducerMessageHandlerTests {
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic1)
.setHeader(KafkaHeaders.TOPIC, topic3)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
@@ -187,7 +211,7 @@ public class KafkaProducerMessageHandlerTests {
handler.handleMessage(message);
ConsumerRecord<Integer, String> record1 = KafkaTestUtils.getSingleRecord(consumer, topic1);
ConsumerRecord<Integer, String> record1 = KafkaTestUtils.getSingleRecord(consumer, topic3);
assertThat(record1).has(key(2));
assertThat(record1).has(partition(1));
assertThat(record1).has(value("foo"));
@@ -198,7 +222,7 @@ public class KafkaProducerMessageHandlerTests {
handler.handleMessage(message);
ConsumerRecord<Integer, String> record2 = KafkaTestUtils.getSingleRecord(consumer, topic1);
ConsumerRecord<Integer, String> record2 = KafkaTestUtils.getSingleRecord(consumer, topic3);
assertThat(record2).has(key(2));
assertThat(record2).has(partition(1));
assertThat(record2).has(value("foo"));
@@ -217,21 +241,20 @@ public class KafkaProducerMessageHandlerTests {
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic1)
.setHeader(KafkaHeaders.TOPIC, topic4)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
handler.handleMessage(message);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic1);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic4);
assertThat(record).has(key(2));
assertThat(record).has(partition(1));
assertThat(record).has(value("foo"));
Message<?> received = successes.receive(10000);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isEqualTo("foo");
// TODO: Change to constant when available
assertThat(received.getHeaders().get("kafka_recordMetadata")).isInstanceOf(RecordMetadata.class);
assertThat(received.getHeaders().get(KafkaHeaders.RECORD_METADATA)).isInstanceOf(RecordMetadata.class);
final RuntimeException fooException = new RuntimeException("foo");
@@ -264,4 +287,94 @@ public class KafkaProducerMessageHandlerTests {
assertThat(((KafkaSendFailureException) received.getPayload()).getRecord()).isNotNull();
}
@Test
public void testOutboundGateway() throws Exception {
ConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka));
ContainerProperties containerProperties = new ContainerProperties(topic6);
final CountDownLatch assigned = new CountDownLatch(1);
containerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// empty
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
assigned.countDown();
}
});
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(producerFactory, container);
template.start();
assertThat(assigned.await(30, TimeUnit.SECONDS)).isTrue();
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
handler.setBeanFactory(mock(BeanFactory.class));
QueueChannel replies = new QueueChannel();
handler.setOutputChannel(replies);
handler.afterPropertiesSet();
Message<?> message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic5)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.build();
handler.handleMessage(message);
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic5);
assertThat(record).has(key(2));
assertThat(record).has(partition(1));
assertThat(record).has(value("foo"));
Map<String, Object> headers = new HashMap<>();
new DefaultKafkaHeaderMapper().toHeaders(record.headers(), headers);
assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic6.getBytes());
ProducerRecord<Integer, String> pr = new ProducerRecord<>(topic6, 0, 1, "FOO", record.headers());
template.send(pr);
Message<?> reply = replies.receive(30_000);
assertThat(reply).isNotNull();
assertThat(reply.getPayload()).isEqualTo("FOO");
assertThat(reply.getHeaders().get(KafkaHeaders.TOPIC)).isNull();
assertThat(reply.getHeaders().get(KafkaHeaders.CORRELATION_ID)).isNull();
message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic5)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.setHeader(KafkaHeaders.REPLY_TOPIC, "bad")
.build();
try {
handler.handleMessage(message);
fail("Expected exception");
}
catch (MessageHandlingException e) {
assertThat(e.getCause().getMessage())
.isEqualTo("The reply topic header [bad] does not match any reply container topic: "
+ "[" + topic6 + "]");
}
message = MessageBuilder.withPayload("foo")
.setHeader(KafkaHeaders.TOPIC, topic5)
.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
.setHeader(KafkaHeaders.PARTITION_ID, 1)
.setHeader(KafkaHeaders.REPLY_PARTITION, 999)
.build();
try {
handler.handleMessage(message);
fail("Expected exception");
}
catch (MessageHandlingException e) {
assertThat(e.getCause().getMessage())
.isEqualTo(
"The reply partition header [999] does not match any reply container partition for topic ["
+ topic6 + "]: [0, 1]");
}
template.stop();
// discard from the test consumer
KafkaTestUtils.getSingleRecord(consumer, topic6);
}
}