GH-9695: Replace AMQP tx-size with batch-size

Fixes: #9695
Issue link: https://github.com/spring-projects/spring-integration/issues/9695

The `txSize` in the `SimpleMessageListenerContainer` has been replaced with more reasonable `batchSize`.
Spring Integration XML support for AMQP module has missed to fix this change: we didn't have a respective test.

* Deprecate `tx-size` (will be removed in `6.5`) XML attribute for the `<amqp:inbound-channel-adapter>`
and introduce `batch-size`
* Cover with the tests
* Fix docs from `tx-size` to `batch-size`

**Auto-cherry-pick to `6.3.x`**
This commit is contained in:
Artem Bilan
2024-12-05 13:58:59 -05:00
parent e2aa904a7b
commit 2cf83b5311
7 changed files with 66 additions and 45 deletions

View File

@@ -59,6 +59,7 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
"receive-timeout", "receive-timeout",
"shutdown-timeout", "shutdown-timeout",
"tx-size", "tx-size",
"batch-size",
"missing-queues-fatal" "missing-queues-fatal"
}; };
@@ -154,7 +155,13 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
} }
builder.addConstructorArgReference(connectionFactoryRef); builder.addConstructorArgReference(connectionFactoryRef);
for (String attributeName : CONTAINER_VALUE_ATTRIBUTES) { for (String attributeName : CONTAINER_VALUE_ATTRIBUTES) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, attributeName); // TODO remove 'tx-size' in 6.5
if ("tx-size".equals(attributeName)) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, attributeName, "batchSize");
}
else {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, attributeName);
}
} }
for (String attributeName : CONTAINER_REFERENCE_ATTRIBUTES) { for (String attributeName : CONTAINER_REFERENCE_ATTRIBUTES) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, attributeName); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, attributeName);

View File

@@ -969,7 +969,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
<xsd:documentation> <xsd:documentation>
Acknowledge Mode for the MessageListenerContainer; default 'AUTO' Acknowledge Mode for the MessageListenerContainer; default 'AUTO'
meaning the adapter automatically acknowledges the message(s) meaning the adapter automatically acknowledges the message(s)
according to the tx-size. according to the batch-size.
</xsd:documentation> </xsd:documentation>
</xsd:annotation> </xsd:annotation>
<xsd:simpleType> <xsd:simpleType>
@@ -1039,7 +1039,7 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
<xsd:documentation> <xsd:documentation>
Specifies how many messages to send to each consumer in a single request. Often this can be set Specifies how many messages to send to each consumer in a single request. Often this can be set
quite high quite high
to improve throughput. It should be greater than or equal to the tx-size value. to improve throughput. It should be greater than or equal to the batch-size value.
</xsd:documentation> </xsd:documentation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>
@@ -1117,10 +1117,23 @@ standard headers to also be mapped. To map all non-standard headers the 'NON_STA
<xsd:annotation> <xsd:annotation>
<xsd:appinfo> <xsd:appinfo>
<xsd:documentation> <xsd:documentation>
[DEPRECATED]
How many messages to process in a single transaction (if the channel is transactional). For best How many messages to process in a single transaction (if the channel is transactional). For best
results it should be results it should be
less than or equal to the prefetch count. less than or equal to the prefetch count.
Not allowed when 'consumers-per-queue' is set. Not allowed when 'consumers-per-queue' is set.
Deprecated in favor of 'batch-size'.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="batch-size" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
How many messages to process in a single request.
For best results it should be less than or equal to the prefetch count.
Not allowed when 'consumers-per-queue' is set.
</xsd:documentation> </xsd:documentation>
</xsd:appinfo> </xsd:appinfo>
</xsd:annotation> </xsd:annotation>

View File

@@ -1,14 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" <beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:amqp="http://www.springframework.org/schema/integration/amqp" xmlns:amqp="http://www.springframework.org/schema/integration/amqp"
xmlns:int="http://www.springframework.org/schema/integration" xmlns:int="http://www.springframework.org/schema/integration"
xmlns:rabbit="http://www.springframework.org/schema/rabbit" xmlns:context="http://www.springframework.org/schema/context"
xmlns:context="http://www.springframework.org/schema/context" xmlns:util="http://www.springframework.org/schema/util"
xmlns:util="http://www.springframework.org/schema/util" xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp https://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
xsi:schemaLocation="http://www.springframework.org/schema/integration/amqp https://www.springframework.org/schema/integration/amqp/spring-integration-amqp.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/rabbit https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd http://www.springframework.org/schema/util https://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd"> http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd">
@@ -20,32 +18,34 @@
</util:properties> </util:properties>
<amqp:inbound-channel-adapter id="rabbitInbound" queue-names="inboundchanneladapter.test.1" <amqp:inbound-channel-adapter id="rabbitInbound" queue-names="inboundchanneladapter.test.1"
batch-mode="EXTRACT_PAYLOADS"/> batch-mode="EXTRACT_PAYLOADS" tx-size="2"/>
<amqp:inbound-channel-adapter id="autoStartFalse" queue-names="inboundchanneladapter.test.2" <amqp:inbound-channel-adapter id="autoStartFalse" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" acknowledge-mode="${ackMode}" missing-queues-fatal="false" /> auto-startup="false" phase="123" acknowledge-mode="${ackMode}"
missing-queues-fatal="false"
batch-size="3"/>
<amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders" <amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders"
channel="requestChannel" queue-names="inboundchanneladapter.test.2" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" auto-startup="false" phase="123"
mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"/> mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"/>
<amqp:inbound-channel-adapter id="withHeaderMapperOnlyCustomHeaders" <amqp:inbound-channel-adapter id="withHeaderMapperOnlyCustomHeaders"
channel="requestChannel" queue-names="inboundchanneladapter.test.2" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" auto-startup="false" phase="123"
mapped-request-headers="foo*"/> mapped-request-headers="foo*"/>
<amqp:inbound-channel-adapter id="withHeaderMapperNothingToMap" <amqp:inbound-channel-adapter id="withHeaderMapperNothingToMap"
channel="requestChannel" queue-names="inboundchanneladapter.test.2" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123" auto-startup="false" phase="123"
mapped-request-headers=""/> mapped-request-headers=""/>
<amqp:inbound-channel-adapter id="withHeaderMapperDefaultMapping" <amqp:inbound-channel-adapter id="withHeaderMapperDefaultMapping"
channel="requestChannel" queue-names="inboundchanneladapter.test.2" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"/> auto-startup="false" phase="123"/>
<amqp:inbound-channel-adapter id="dmlc" queue-names="inboundchanneladapter.test.2" consumers-per-queue="2" <amqp:inbound-channel-adapter id="dmlc" queue-names="inboundchanneladapter.test.2" consumers-per-queue="2"
auto-startup="false" /> auto-startup="false"/>
<int:channel id="requestChannel"> <int:channel id="requestChannel">
<int:queue/> <int:queue/>

View File

@@ -1,5 +1,5 @@
/* /*
* Copyright 2002-2022 the original author or authors. * Copyright 2002-2024 the original author or authors.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -16,8 +16,7 @@
package org.springframework.integration.amqp.config; package org.springframework.integration.amqp.config;
import org.junit.Test; import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message; import org.springframework.amqp.core.Message;
@@ -37,10 +36,10 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/** /**
* @author Mark Fisher * @author Mark Fisher
@@ -49,8 +48,7 @@ import static org.assertj.core.api.Assertions.assertThat;
* *
* @since 2.1 * @since 2.1
*/ */
@ContextConfiguration @SpringJUnitConfig
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext @DirtiesContext
public class AmqpInboundChannelAdapterParserTests { public class AmqpInboundChannelAdapterParserTests {
@@ -71,6 +69,8 @@ public class AmqpInboundChannelAdapterParserTests {
.isInstanceOf(SimpleMessageListenerContainer.class); .isInstanceOf(SimpleMessageListenerContainer.class);
assertThat(TestUtils.getPropertyValue(adapter, "batchMode", BatchMode.class)) assertThat(TestUtils.getPropertyValue(adapter, "batchMode", BatchMode.class))
.isEqualTo(BatchMode.EXTRACT_PAYLOADS); .isEqualTo(BatchMode.EXTRACT_PAYLOADS);
assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer.batchSize", Integer.class))
.isEqualTo(2);
} }
@Test @Test
@@ -95,6 +95,8 @@ public class AmqpInboundChannelAdapterParserTests {
.isEqualTo(AcknowledgeMode.NONE); .isEqualTo(AcknowledgeMode.NONE);
assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class)) assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer.missingQueuesFatal", Boolean.class))
.isFalse(); .isFalse();
assertThat(TestUtils.getPropertyValue(adapter, "messageListenerContainer.batchSize", Integer.class))
.isEqualTo(3);
} }
@Test @Test
@@ -216,14 +218,13 @@ public class AmqpInboundChannelAdapterParserTests {
@Test @Test
public void testInt2971HeaderMapperAndMappedHeadersExclusivity() { public void testInt2971HeaderMapperAndMappedHeadersExclusivity() {
try { assertThatExceptionOfType(BeanDefinitionParsingException.class)
new ClassPathXmlApplicationContext("AmqpInboundChannelAdapterParserTests-headerMapper-fail-context.xml", .isThrownBy(() ->
this.getClass()).close(); new ClassPathXmlApplicationContext(
} "AmqpInboundChannelAdapterParserTests-headerMapper-fail-context.xml",
catch (BeanDefinitionParsingException e) { getClass()))
assertThat(e.getMessage().startsWith("Configuration problem: The 'header-mapper' attribute " + .withMessageStartingWith("Configuration problem: The 'header-mapper' attribute " +
"is mutually exclusive with 'mapped-request-headers' or 'mapped-reply-headers'")).isTrue(); "is mutually exclusive with 'mapped-request-headers' or 'mapped-reply-headers'");
}
} }
} }

View File

@@ -32,7 +32,7 @@ The following adapters are available:
* xref:amqp/inbound-gateway.adoc[Inbound Gateway] * xref:amqp/inbound-gateway.adoc[Inbound Gateway]
* xref:amqp/outbound-channel-adapter.adoc[Outbound Channel Adapter] * xref:amqp/outbound-channel-adapter.adoc[Outbound Channel Adapter]
* xref:amqp/outbound-gateway.adoc[Outbound Gateway] * xref:amqp/outbound-gateway.adoc[Outbound Gateway]
* xref:amqp-async-outbound-gateway[Async Outbound Gateway] * xref:amqp/async-outbound-gateway.adoc[Async Outbound Gateway]
* xref:amqp/rmq-streams.adoc#rmq-stream-inbound-channel-adapter[RabbitMQ Stream Queue Inbound Channel Adapter] * xref:amqp/rmq-streams.adoc#rmq-stream-inbound-channel-adapter[RabbitMQ Stream Queue Inbound Channel Adapter]
* xref:amqp/rmq-streams.adoc#rmq-stream-outbound-channel-adapter[RabbitMQ Stream Queue Outbound Channel Adapter] * xref:amqp/rmq-streams.adoc#rmq-stream-outbound-channel-adapter[RabbitMQ Stream Queue Outbound Channel Adapter]

View File

@@ -87,7 +87,7 @@ XML::
task-executor="" <22> task-executor="" <22>
transaction-attribute="" <23> transaction-attribute="" <23>
transaction-manager="" <24> transaction-manager="" <24>
tx-size="" <25> batch-size="" <25>
consumers-per-queue <26> consumers-per-queue <26>
batch-mode="MESSAGES"/> <27> batch-mode="MESSAGES"/> <27>
@@ -146,7 +146,7 @@ By default, this value is `Integer.MAX_VALUE`, meaning that this container start
Optional. Optional.
<17> Tells the AMQP broker how many messages to send to each consumer in a single request. <17> Tells the AMQP broker how many messages to send to each consumer in a single request.
Often, you can set this value high to improve throughput. Often, you can set this value high to improve throughput.
It should be greater than or equal to the transaction size (see the `tx-size` attribute, later in this list). It should be greater than or equal to the transaction size (see the `batch-size` attribute, later in this list).
Optional (defaults to `1`). Optional (defaults to `1`).
<18> Receive timeout in milliseconds. <18> Receive timeout in milliseconds.
Optional (defaults to `1000`). Optional (defaults to `1000`).
@@ -173,7 +173,7 @@ If the `channelTransacted` flag is `false`, no transaction semantics apply to th
For further information, see For further information, see
https://docs.spring.io/spring-amqp/reference/html/%255Freference.html#%5Ftransactions[Transactions with Spring AMQP]. https://docs.spring.io/spring-amqp/reference/html/%255Freference.html#%5Ftransactions[Transactions with Spring AMQP].
Optional. Optional.
<25> Tells the `SimpleMessageListenerContainer` how many messages to process in a single transaction (if the channel is transactional). <25> Tells the `SimpleMessageListenerContainer` how many messages to process in a single request.
For best results, it should be less than or equal to the value set in `prefetch-count`. For best results, it should be less than or equal to the value set in `prefetch-count`.
Not allowed when 'consumers-per-queue' is set. Not allowed when 'consumers-per-queue' is set.
Optional (defaults to `1`). Optional (defaults to `1`).

View File

@@ -10,7 +10,7 @@ Java DSL::
+ +
[source, java, role="primary"] [source, java, role="primary"]
---- ----
@Bean // return the upper cased payload @Bean // return the upper-cased payload
public IntegrationFlow amqpInboundGateway(ConnectionFactory connectionFactory) { public IntegrationFlow amqpInboundGateway(ConnectionFactory connectionFactory) {
return IntegrationFlow.from(Amqp.inboundGateway(connectionFactory, "foo")) return IntegrationFlow.from(Amqp.inboundGateway(connectionFactory, "foo"))
.transform(String.class, String::toUpperCase) .transform(String.class, String::toUpperCase)