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

GH-3966: Kafka XML config: Expose more attributes

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3966

Exposed setters in the `KafkaInboundGateway` and `KafkaMessageDrivenChannelAdapter`
as an XML attributes for `kafka` namespace

**Cherry-pick to `5.5.x`**
This commit is contained in:
abilan
2022-12-12 15:03:45 -05:00
committed by Gary Russell
parent 280eb29f62
commit e63edde7cb
7 changed files with 128 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019 the original author or authors.
* Copyright 2019-2022 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.
@@ -27,6 +27,7 @@ import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
* Inbound gateway parser.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 5.4
*
@@ -48,6 +49,9 @@ public class KafkaInboundGatewayParser extends AbstractInboundGatewayParser {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
"on-partitions-assigned-seek-callback");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "bind-source-record");
}
@Override

8
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2019 the original author or authors.
* Copyright 2015-2022 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.
@@ -59,6 +59,12 @@ public class KafkaMessageDrivenChannelAdapterParser extends AbstractChannelAdapt
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-message-strategy");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "retry-template");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "recovery-callback");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "record-filter-strategy");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
"on-partitions-assigned-seek-callback");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ack-discarded");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "filter-in-retry");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "bind-source-record");
return builder.getBeanDefinition();
}

64
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka.xsd
View File

@@ -373,6 +373,46 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="record-filter-strategy" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Wrap a 'KafkaMessageDrivenChannelAdapter.IntegrationRecordMessageListener'
into 'FilteringMessageListenerAdapter' with the provided 'RecordFilterStrategy'.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.kafka.listener.adapter.RecordFilterStrategy" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="ack-discarded">
<xsd:annotation>
<xsd:documentation>
A boolean flag to indicate if 'FilteringMessageListenerAdapter
should acknowledge discarded records or not.
Does not make sense if 'record-filter-strategy' isn't specified.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="filter-in-retry">
<xsd:annotation>
<xsd:documentation>
A boolean flag to specify the order in which the filter and retry
operations are performed.
Does not make sense if only one of 'retry-template' or
'record-filter-strategy' is present, or none.
When true, the filter is called for each retry; when false, the filter is only
called once for each delivery from the container.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -725,6 +765,30 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="on-partitions-assigned-seek-callback" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A 'BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback>'
bean reference for seeks management.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.function.BiConsumer" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="bind-source-record">
<xsd:annotation>
<xsd:documentation>
Set to true to bind the source consumer record in the header named
'IntegrationMessageHeaderAccessor#SOURCE_DATA'.
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attributeGroup ref="errorMessageStrategyGroup" />
</xsd:complexType>

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests-context.xml
View File

@@ -19,7 +19,9 @@
payload-type="java.lang.String"
error-message-strategy="ems"
retry-template="retryTemplate"
recovery-callback="recoveryCallback"/>
recovery-callback="recoveryCallback"
bind-source-record="true"
on-partitions-assigned-seek-callback="onPartitionsAssignedSeekCallback"/>
<bean id="template" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.kafka.core.KafkaTemplate"/>
@@ -44,6 +46,10 @@
</constructor-arg>
</bean>
<bean id="onPartitionsAssignedSeekCallback" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="java.util.function.BiConsumer"/>
</bean>
<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy"/>
<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"/>

10
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaInboundGatewayTests.java
View File

@@ -30,6 +30,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 5.4
*
@@ -58,13 +59,16 @@ public class KafkaInboundGatewayTests {
assertThat(TestUtils.getPropertyValue(this.gateway1, "listener.fallbackType"))
.isEqualTo(String.class);
assertThat(TestUtils.getPropertyValue(this.gateway1, "errorMessageStrategy"))
.isSameAs(this.context.getBean("ems"));
.isSameAs(this.context.getBean("ems"));
assertThat(TestUtils.getPropertyValue(this.gateway1, "retryTemplate"))
.isSameAs(this.context.getBean("retryTemplate"));
.isSameAs(this.context.getBean("retryTemplate"));
assertThat(TestUtils.getPropertyValue(this.gateway1, "recoveryCallback"))
.isSameAs(this.context.getBean("recoveryCallback"));
.isSameAs(this.context.getBean("recoveryCallback"));
assertThat(TestUtils.getPropertyValue(this.gateway1, "onPartitionsAssignedSeekCallback"))
.isSameAs(this.context.getBean("onPartitionsAssignedSeekCallback"));
assertThat(TestUtils.getPropertyValue(this.gateway1, "messagingTemplate.sendTimeout")).isEqualTo(5000L);
assertThat(TestUtils.getPropertyValue(this.gateway1, "replyTimeout")).isEqualTo(43L);
assertThat(TestUtils.getPropertyValue(this.gateway1, "bindSourceRecord", Boolean.class)).isTrue();
}
}

39
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml
View File

@@ -1,9 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">
@@ -20,7 +20,12 @@
payload-type="java.lang.String"
error-message-strategy="ems"
retry-template="retryTemplate"
recovery-callback="recoveryCallback" />
recovery-callback="recoveryCallback"
bind-source-record="true"
on-partitions-assigned-seek-callback="onPartitionsAssignedSeekCallback"
filter-in-retry="true"
ack-discarded="true"
record-filter-strategy="recordFilterStrategy"/>
<int-kafka:message-driven-channel-adapter
id="kafkaBatchListener"
@@ -31,7 +36,7 @@
channel="nullChannel"
mode="batch"
message-converter="messageConverter"
error-channel="errorChannel" />
error-channel="errorChannel"/>
<bean id="messageConverter" class="org.springframework.kafka.support.converter.MessagingMessageConverter"/>
@@ -40,39 +45,47 @@
<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>
<entry key="" value="" />
<entry key="" value=""/>
</map>
</constructor-arg>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.kafka.listener.ContainerProperties">
<constructor-arg name="topics" value="foo" />
<constructor-arg name="topics" value="foo"/>
</bean>
</constructor-arg>
</bean>
<bean id="onPartitionsAssignedSeekCallback" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="java.util.function.BiConsumer"/>
</bean>
<bean id="recordFilterStrategy" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.kafka.listener.adapter.RecordFilterStrategy"/>
</bean>
<bean id="container2" class="org.springframework.kafka.listener.KafkaMessageListenerContainer">
<constructor-arg>
<bean class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
<constructor-arg>
<map>
<entry key="" value="" />
<entry key="" value=""/>
</map>
</constructor-arg>
</bean>
</constructor-arg>
<constructor-arg>
<bean class="org.springframework.kafka.listener.ContainerProperties">
<constructor-arg name="topics" value="foo" />
<constructor-arg name="topics" value="foo"/>
</bean>
</constructor-arg>
</bean>
<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy" />
<bean id="ems" class="org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy"/>
<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate" />
<bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate"/>
<bean id="recoveryCallback" class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer" />
<bean id="recoveryCallback" class="org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer"/>
</beans>

13
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java
View File

@@ -24,6 +24,7 @@ import java.util.Collections;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -43,7 +44,7 @@ import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* @author Artem Bilan.
* @author Artem Bilan
* @author Gary Russell
*
* @since 5.4
@@ -52,6 +53,9 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
@DirtiesContext
class KafkaMessageDrivenChannelAdapterParserTests {
@Autowired
private ApplicationContext context;
@Autowired
private NullChannel nullChannel;
@@ -92,6 +96,13 @@ class KafkaMessageDrivenChannelAdapterParserTests {
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "errorMessageStrategy")).isSameAs(this.ems);
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "retryTemplate")).isSameAs(this.retryTemplate);
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "recoveryCallback")).isSameAs(this.recoveryCallback);
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "onPartitionsAssignedSeekCallback"))
.isSameAs(this.context.getBean("onPartitionsAssignedSeekCallback"));
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "bindSourceRecord", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "filterInRetry", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "ackDiscarded", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(this.kafkaListener, "recordFilterStrategy"))
.isSameAs(this.context.getBean("recordFilterStrategy"));
}
@Test