diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java index a98c1a60c186f..fe48be13475ec 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/DefaultJmsMessageListenerContainer.java @@ -161,4 +161,15 @@ protected void stopSharedConnection() { } super.stopSharedConnection(); } + + /** + * Forces the listener container to refresh its connection and recreate consumers, which will re-resolve the + * temporary reply destination. Used when a pending reply-destination refresh must be consumed but cached consumers + * would otherwise never call the destination resolver again. + */ + public void recoverReplyDestinationAfterRefresh() { + if (isRunning() && !isRecovering()) { + recoverAfterListenerSetupFailure(); + } + } } diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java index 6f5b4ef9275fa..9d4301131107c 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManager.java @@ -16,7 +16,10 @@ */ package org.apache.camel.component.jms.reply; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -40,6 +43,7 @@ import org.apache.camel.support.service.ServiceSupport; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.SimpleMessageListenerContainer; import org.springframework.jms.support.destination.DestinationResolver; /** @@ -48,6 +52,8 @@ public class TemporaryQueueReplyManager extends ReplyManagerSupport { final TemporaryReplyQueueDestinationResolver destinationResolver; + private ExecutorService refreshRecoveryExecutor; + private final AtomicBoolean refreshRecoveryScheduled = new AtomicBoolean(); public TemporaryQueueReplyManager(CamelContext camelContext, TemporaryQueueResolver resolver) { super(camelContext); @@ -57,9 +63,95 @@ public TemporaryQueueReplyManager(CamelContext camelContext, TemporaryQueueResol @Override protected void doStop() throws Exception { super.doStop(); + if (refreshRecoveryExecutor != null) { + camelContext.getExecutorServiceManager().shutdownNow(refreshRecoveryExecutor); + refreshRecoveryExecutor = null; + } ServiceHelper.stopService(destinationResolver); } + private void triggerReplyDestinationRecovery() { + if (listenerContainer == null || isStopping() || isStopped()) { + return; + } + if (!destinationResolver.isRefreshPending()) { + return; + } + if (!refreshRecoveryScheduled.compareAndSet(false, true)) { + return; + } + try { + getRefreshRecoveryExecutor().execute(this::runReplyDestinationRecovery); + } catch (RejectedExecutionException e) { + refreshRecoveryScheduled.set(false); + } + } + + private void runReplyDestinationRecovery() { + try { + long delay = endpoint.getRecoveryInterval() >= 0 ? endpoint.getRecoveryInterval() : 5000L; + if (!sleepQuietly(delay)) { + return; + } + int attempts = 0; + while (destinationResolver.isRefreshPending() && !isStopping() && !isStopped() + && listenerContainer != null && listenerContainer.isRunning() && attempts < 20) { + try { + if (listenerContainer instanceof DefaultJmsMessageListenerContainer dmlc) { + if (dmlc.isRecovering()) { + if (!sleepQuietly(delay)) { + return; + } + attempts++; + continue; + } + dmlc.recoverReplyDestinationAfterRefresh(); + } else if (listenerContainer instanceof SimpleMessageListenerContainer smlc) { + smlc.stop(); + smlc.start(); + } else { + listenerContainer.stop(); + listenerContainer.start(); + } + } catch (Exception e) { + log.warn("Failed to trigger recovery of temporary reply destination on endpoint: {}", + endpoint.getEndpointUri(), e); + } + if (!destinationResolver.isRefreshPending()) { + break; + } + if (!sleepQuietly(delay)) { + return; + } + attempts++; + } + } finally { + refreshRecoveryScheduled.set(false); + if (destinationResolver.isRefreshPending() && !isStopping() && !isStopped() + && listenerContainer != null && listenerContainer.isRunning()) { + triggerReplyDestinationRecovery(); + } + } + } + + private boolean sleepQuietly(long millis) { + try { + Thread.sleep(millis); + return !isStopping() && !isStopped(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + private ExecutorService getRefreshRecoveryExecutor() { + if (refreshRecoveryExecutor == null) { + String name = "JmsTemporaryReplyToRefresh[" + endpoint.getDestinationName() + "]"; + refreshRecoveryExecutor = camelContext.getExecutorServiceManager().newSingleThreadExecutor(this, name); + } + return refreshRecoveryExecutor; + } + @Override protected ReplyHandler createReplyHandler( ReplyManager replyManager, Exchange exchange, AsyncCallback callback, @@ -283,51 +375,76 @@ final class TemporaryReplyQueueDestinationResolver extends ServiceSupport // the destination, it would deadlock trying to acquire BaseService.lock. private final Lock destinationLock = new ReentrantLock(); private volatile TemporaryQueue queue; - private final AtomicBoolean refreshWanted = new AtomicBoolean(); + private final AtomicLong refreshGeneration = new AtomicLong(); + private volatile long publishedGeneration; private final TemporaryQueueResolver custom; public TemporaryReplyQueueDestinationResolver(TemporaryQueueResolver custom) { this.custom = custom; } + boolean isRefreshPending() { + return refreshGeneration.get() != publishedGeneration; + } + @Override public Destination resolveDestinationName(Session session, String destinationName, boolean pubSubDomain) throws JMSException { - // fast path: queue already resolved and no refresh needed - TemporaryQueue answer = queue; - if (answer != null && !refreshWanted.get()) { - return answer; - } destinationLock.lock(); try { - if (queue == null || refreshWanted.get()) { - refreshWanted.set(false); - if (custom != null) { - if (queue != null) { - try { - custom.delete(queue); - } catch (Exception e) { - // ignore - } + TemporaryQueue answer = queue; + if (answer != null && !isRefreshPending()) { + return answer; + } + long generationToHandle = refreshGeneration.get(); + TemporaryQueue previousQueue = queue; + if (previousQueue != null) { + try { + if (custom != null) { + custom.delete(previousQueue); + } else { + previousQueue.delete(); } - queue = custom.createTemporaryQueue(session); - } else { - queue = session.createTemporaryQueue(); + } catch (Exception e) { + // ignore } - setReplyTo(queue); + queue = null; + } + TemporaryQueue refreshedQueue; + if (custom != null) { + refreshedQueue = custom.createTemporaryQueue(session); + } else { + refreshedQueue = session.createTemporaryQueue(); + } + if (refreshGeneration.get() == generationToHandle) { + queue = refreshedQueue; + setReplyTo(refreshedQueue); + publishedGeneration = generationToHandle; if (log.isDebugEnabled()) { - log.debug("Refreshed Temporary ReplyTo Queue. New queue: {}", queue.getQueueName()); + log.debug("Refreshed Temporary ReplyTo Queue. New queue: {}", refreshedQueue.getQueueName()); + } + return refreshedQueue; + } + // a newer refresh was requested while creating the queue; discard this attempt + try { + if (custom != null) { + custom.delete(refreshedQueue); + } else { + refreshedQueue.delete(); } + } catch (Exception e) { + // ignore } + return null; } finally { destinationLock.unlock(); } - return queue; } public void scheduleRefresh() { - refreshWanted.set(true); + refreshGeneration.incrementAndGet(); replyTo = null; + triggerReplyDestinationRecovery(); } @Override @@ -346,6 +463,7 @@ protected void doStop() throws Exception { } queue = null; } + publishedGeneration = refreshGeneration.get(); } finally { destinationLock.unlock(); } diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/JmsTemporaryReplyToRequestReplyIT.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/JmsTemporaryReplyToRequestReplyIT.java new file mode 100644 index 0000000000000..880a03c4a394b --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/JmsTemporaryReplyToRequestReplyIT.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.camel.component.jms.reply; + +import org.apache.camel.CamelContext; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jms.JmsComponent; +import org.apache.camel.test.infra.artemis.common.ConnectionFactoryHelper; +import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory; +import org.apache.camel.test.infra.core.DefaultCamelContextExtension; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge; +import static org.assertj.core.api.Assertions.assertThat; + +class JmsTemporaryReplyToRequestReplyIT { + + private static final String REQUEST_QUEUE = "JmsTemporaryReplyToRequestReplyIT.request"; + + @RegisterExtension + static ArtemisService service = ArtemisServiceFactory.createVMService(); + + @RegisterExtension + static DefaultCamelContextExtension contextExtension = new DefaultCamelContextExtension(); + + private ProducerTemplate template; + + @BeforeEach + void setUp() throws Exception { + CamelContext context = contextExtension.getContext(); + JmsComponent component = jmsComponentAutoAcknowledge(ConnectionFactoryHelper.createConnectionFactory(service)); + context.addComponent("jms", component); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("jms:queue:" + REQUEST_QUEUE).routeId("server") + .transform(simple("echo:${body}")); + } + }); + template = contextExtension.getProducerTemplate(); + } + + @Test + void shouldSupportConsecutiveTemporaryReplyRequests() { + assertThat(template.requestBody("jms:queue:" + REQUEST_QUEUE, "first", String.class)) + .isEqualTo("echo:first"); + assertThat(template.requestBody("jms:queue:" + REQUEST_QUEUE, "second", String.class)) + .isEqualTo("echo:second"); + } +} diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManagerRefreshTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManagerRefreshTest.java new file mode 100644 index 0000000000000..36b1f679a3cfe --- /dev/null +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/reply/TemporaryQueueReplyManagerRefreshTest.java @@ -0,0 +1,256 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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.apache.camel.component.jms.reply; + +import java.util.concurrent.atomic.AtomicInteger; + +import jakarta.jms.Connection; +import jakarta.jms.JMSException; +import jakarta.jms.Session; +import jakarta.jms.TemporaryQueue; + +import org.apache.camel.CamelContext; +import org.apache.camel.component.jms.JmsConfiguration; +import org.apache.camel.component.jms.JmsEndpoint; +import org.apache.camel.component.jms.TemporaryQueueResolver; +import org.apache.camel.test.infra.artemis.common.ConnectionFactoryHelper; +import org.apache.camel.test.infra.artemis.services.ArtemisService; +import org.apache.camel.test.infra.artemis.services.ArtemisServiceFactory; +import org.apache.camel.test.infra.core.DefaultCamelContextExtension; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class TemporaryQueueReplyManagerRefreshTest { + + @RegisterExtension + static DefaultCamelContextExtension contextExtension = new DefaultCamelContextExtension(); + + @RegisterExtension + static ArtemisService service = ArtemisServiceFactory.createVMService(); + + private TemporaryQueueReplyManager replyManager; + + @BeforeEach + void setUp() { + CamelContext context = contextExtension.getContext(); + JmsEndpoint endpoint = mock(JmsEndpoint.class); + when(endpoint.getCamelContext()).thenReturn(context); + when(endpoint.getConfiguration()).thenReturn(new JmsConfiguration()); + when(endpoint.getDestinationName()).thenReturn("TemporaryQueueReplyManagerRefreshTest"); + when(endpoint.getEndpointUri()).thenReturn("jms:queue:TemporaryQueueReplyManagerRefreshTest"); + + replyManager = new TemporaryQueueReplyManager(context, null); + replyManager.setEndpoint(endpoint); + } + + @Test + void shouldRetryAfterFailedTemporaryQueueCreation() throws Exception { + FailOnSecondCreateResolver resolver = new FailOnSecondCreateResolver(); + replyManager = new TemporaryQueueReplyManager(contextExtension.getContext(), resolver); + replyManager.setEndpoint(createEndpoint()); + + try (Connection connection = ConnectionFactoryHelper.createConnectionFactory(service).createConnection()) { + connection.start(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver + = replyManager.destinationResolver; + + destinationResolver.resolveDestinationName(session, "temporary", false); + assertThat(replyManager.getReplyTo()).isNotNull(); + assertThat(resolver.createAttempts.get()).isEqualTo(1); + + destinationResolver.scheduleRefresh(); + assertThat(destinationResolver.isRefreshPending()).isTrue(); + assertThat(replyManager.getReplyTo()).isNull(); + + assertThatThrownBy(() -> destinationResolver.resolveDestinationName(session, "temporary", false)) + .isInstanceOf(JMSException.class); + assertThat(destinationResolver.isRefreshPending()).isTrue(); + assertThat(replyManager.getReplyTo()).isNull(); + assertThat(resolver.createAttempts.get()).isEqualTo(2); + + destinationResolver.resolveDestinationName(session, "temporary", false); + assertThat(destinationResolver.isRefreshPending()).isFalse(); + assertThat(replyManager.getReplyTo()).isNotNull(); + assertThat(resolver.createAttempts.get()).isEqualTo(3); + } + } + + @Test + void shouldPublishLatestRefreshGeneration() throws Exception { + CountingTemporaryQueueResolver resolver = new CountingTemporaryQueueResolver(); + replyManager = new TemporaryQueueReplyManager(contextExtension.getContext(), resolver); + replyManager.setEndpoint(createEndpoint()); + + try (Connection connection = ConnectionFactoryHelper.createConnectionFactory(service).createConnection()) { + connection.start(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver + = replyManager.destinationResolver; + + TemporaryQueue firstQueue + = (TemporaryQueue) destinationResolver.resolveDestinationName(session, "temporary", false); + assertThat(replyManager.getReplyTo()).isEqualTo(firstQueue); + + destinationResolver.scheduleRefresh(); + destinationResolver.scheduleRefresh(); + + TemporaryQueue resolvedQueue + = (TemporaryQueue) destinationResolver.resolveDestinationName(session, "temporary", false); + + assertThat(destinationResolver.isRefreshPending()).isFalse(); + assertThat(replyManager.getReplyTo()).isEqualTo(resolvedQueue); + assertThat(resolvedQueue).isNotEqualTo(firstQueue); + assertThat(resolver.createAttempts.get()).isEqualTo(2); + } + } + + @Test + void shouldKeepRefreshPendingWhenCreationKeepsFailing() throws Exception { + AlwaysFailingTemporaryQueueResolver resolver = new AlwaysFailingTemporaryQueueResolver(); + replyManager = new TemporaryQueueReplyManager(contextExtension.getContext(), resolver); + replyManager.setEndpoint(createEndpoint()); + + try (Connection connection = ConnectionFactoryHelper.createConnectionFactory(service).createConnection()) { + connection.start(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver + = replyManager.destinationResolver; + + destinationResolver.scheduleRefresh(); + assertThatThrownBy(() -> destinationResolver.resolveDestinationName(session, "temporary", false)) + .isInstanceOf(JMSException.class); + + assertThat(destinationResolver.isRefreshPending()).isTrue(); + assertThat(replyManager.getReplyTo()).isNull(); + } + } + + @Test + void shouldDiscardPublishWhenRefreshGenerationChangesDuringResolve() throws Exception { + InterleavingRefreshResolver resolver = new InterleavingRefreshResolver(); + replyManager = new TemporaryQueueReplyManager(contextExtension.getContext(), resolver); + replyManager.setEndpoint(createEndpoint()); + resolver.attach(replyManager.destinationResolver); + + try (Connection connection = ConnectionFactoryHelper.createConnectionFactory(service).createConnection()) { + connection.start(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + + TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver + = replyManager.destinationResolver; + + destinationResolver.resolveDestinationName(session, "temporary", false); + assertThat(replyManager.getReplyTo()).isNotNull(); + + destinationResolver.scheduleRefresh(); + assertThat(destinationResolver.resolveDestinationName(session, "temporary", false)).isNull(); + assertThat(destinationResolver.isRefreshPending()).isTrue(); + assertThat(replyManager.getReplyTo()).isNull(); + + TemporaryQueue resolvedQueue + = (TemporaryQueue) destinationResolver.resolveDestinationName(session, "temporary", false); + assertThat(resolvedQueue).isNotNull(); + assertThat(destinationResolver.isRefreshPending()).isFalse(); + assertThat(replyManager.getReplyTo()).isEqualTo(resolvedQueue); + } + } + + private JmsEndpoint createEndpoint() { + JmsEndpoint endpoint = mock(JmsEndpoint.class); + when(endpoint.getCamelContext()).thenReturn(contextExtension.getContext()); + when(endpoint.getConfiguration()).thenReturn(new JmsConfiguration()); + when(endpoint.getDestinationName()).thenReturn("TemporaryQueueReplyManagerRefreshTest"); + when(endpoint.getEndpointUri()).thenReturn("jms:queue:TemporaryQueueReplyManagerRefreshTest"); + return endpoint; + } + + private static final class FailOnSecondCreateResolver implements TemporaryQueueResolver { + private final AtomicInteger createAttempts = new AtomicInteger(); + + @Override + public TemporaryQueue createTemporaryQueue(Session session) throws JMSException { + if (createAttempts.incrementAndGet() == 2) { + throw new JMSException("simulated broker failure"); + } + return session.createTemporaryQueue(); + } + + @Override + public void delete(TemporaryQueue queue) { + // noop + } + } + + private static final class CountingTemporaryQueueResolver implements TemporaryQueueResolver { + private final AtomicInteger createAttempts = new AtomicInteger(); + + @Override + public TemporaryQueue createTemporaryQueue(Session session) throws JMSException { + createAttempts.incrementAndGet(); + return session.createTemporaryQueue(); + } + + @Override + public void delete(TemporaryQueue queue) { + // noop + } + } + + private static final class AlwaysFailingTemporaryQueueResolver implements TemporaryQueueResolver { + @Override + public TemporaryQueue createTemporaryQueue(Session session) throws JMSException { + throw new JMSException("simulated broker failure"); + } + + @Override + public void delete(TemporaryQueue queue) { + // noop + } + } + + private static final class InterleavingRefreshResolver implements TemporaryQueueResolver { + private TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver; + private final AtomicInteger createAttempts = new AtomicInteger(); + + void attach(TemporaryQueueReplyManager.TemporaryReplyQueueDestinationResolver destinationResolver) { + this.destinationResolver = destinationResolver; + } + + @Override + public TemporaryQueue createTemporaryQueue(Session session) throws JMSException { + if (createAttempts.incrementAndGet() == 2) { + destinationResolver.scheduleRefresh(); + } + return session.createTemporaryQueue(); + } + + @Override + public void delete(TemporaryQueue queue) { + // noop + } + } +} diff --git a/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties b/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties index e230d8d32d785..378f1ef4adbc0 100644 --- a/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties +++ b/core/camel-core-model/src/generated/resources/META-INF/services/org/apache/camel/model.properties @@ -13,11 +13,16 @@ beanFactory beanio bearerToken bindy +blacklistServiceFilter +cachingServiceDiscovery cbor choice circuitBreaker claimCheck +combinedServiceDiscovery +combinedServiceFilter constant +consulServiceDiscovery contextScan convertBodyTo convertHeaderTo @@ -27,6 +32,7 @@ csimple csv custom customLoadBalancer +customServiceFilter customTransformer customValidator dataFormatTransformer @@ -34,9 +40,11 @@ dataFormats datasonnet deadLetterChannel defaultErrorHandler +defaultLoadBalancer delay delete dfdl +dnsServiceDiscovery doCatch doFinally doTry @@ -65,6 +73,7 @@ groovyXml gzipDeflater head header +healthyServiceFilter hl7 hl7terser ical @@ -86,6 +95,7 @@ jsonApi jsonpath jtaTransactionErrorHandler kamelet +kubernetesServiceDiscovery langChain4jCharacterTokenizer langChain4jLineTokenizer langChain4jParagraphTokenizer @@ -93,6 +103,7 @@ langChain4jSentenceTokenizer langChain4jWordTokenizer language loadBalance +loadBalancerConfiguration loadTransformer log loop @@ -119,6 +130,7 @@ outputType packageScan param parquetAvro +passThroughServiceFilter patch pausable pgp @@ -175,6 +187,12 @@ sample script security securityDefinitions +serviceCall +serviceCallConfiguration +serviceChooserConfiguration +serviceDiscoveryConfiguration +serviceExpression +serviceFilterConfiguration setBody setExchangePattern setHeader @@ -189,6 +207,7 @@ sort spel split springTransactionErrorHandler +staticServiceDiscovery step stickyLoadBalancer stop @@ -206,6 +225,7 @@ threads thrift throttle throwException +tidyMarkup to toD tokenize @@ -215,6 +235,7 @@ transacted transform transformDataType transformers +typeFilter ubl univocityCsv univocityFixed @@ -228,6 +249,7 @@ variable wasm weightedLoadBalancer when +whenSkipSendToEndpoint wireTap xmlSecurity xpath @@ -236,3 +258,4 @@ xtokenize yaml zipDeflater zipFile +zookeeperServiceDiscovery