This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 68329f0919bc CAMEL-25684: camel-slack - the consumer keeps the
messages of a poll it did not route when the route is suspended
68329f0919bc is described below
commit 68329f0919bc2b7f22a6ff2ecfc373b58269948c
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 16 20:29:09 2026 +0530
CAMEL-25684: camel-slack - the consumer keeps the messages of a poll it did
not route when the route is suspended
[CAMEL-25684](https://issues.apache.org/jira/browse/CAMEL-25684)
`SlackConsumer.createExchanges` moved the timestamp to the most recent
message of the poll before `processBatch` routes the messages. `processBatch`
stops when `isBatchAllowed()` is false, which `ScheduledBatchPollingConsumer`
returns when the route is being suspended (`RouteController.suspendRoute`, also
from a route policy that suspends the route through the route controller, such
as `DurationRoutePolicy`, or `CamelContext.suspend()`, which resumes the same
consumer). The rest of th [...]
Closes #27923
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/slack/SlackConsumer.java | 44 ++-
...ckConsumerSuspendedBatchPooledExchangeTest.java | 74 +++++
.../slack/SlackConsumerSuspendedBatchTest.java | 311 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
4 files changed, 430 insertions(+), 7 deletions(-)
diff --git
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
index 216534e61dd5..1b9ad2b628bf 100644
---
a/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
+++
b/components/camel-slack/src/main/java/org/apache/camel/component/slack/SlackConsumer.java
@@ -53,6 +53,9 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
private final SlackEndpoint slackEndpoint;
private Slack slack;
private String timestamp;
+ // messages of a previous poll, the most recent first, that were not
routed because the batch ended early. The
+ // timestamp is already past them, so they are routed with the messages of
the next poll
+ private final List<Message> notRouted = new ArrayList<>();
private String channelId;
public SlackConsumer(SlackEndpoint endpoint, Processor processor) {
@@ -120,6 +123,18 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
}
}
+ if (ObjectHelper.isNotEmpty(messages)) {
+ // the most recent message comes first
+ timestamp = messages.get(0).getTs();
+ }
+ if (!notRouted.isEmpty()) {
+ // they are older than the messages of this poll
+ List<Message> all = messages != null ? new ArrayList<>(messages) :
new ArrayList<>();
+ all.addAll(notRouted);
+ notRouted.clear();
+ messages = all;
+ }
+
Queue<Exchange> exchanges = createExchanges(messages);
return processBatch(CastUtils.cast(exchanges));
}
@@ -148,18 +163,12 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
if (slackEndpoint.isNaturalOrder()) {
for (int i = list.size() - 1; i >= 0; i--) {
Message message = list.get(i);
- if (i == 0) {
- timestamp = message.getTs();
- }
Exchange exchange = createExchange(message);
answer.add(exchange);
}
} else {
for (int i = 0; i < list.size(); i++) {
Message message = list.get(i);
- if (i == 0) {
- timestamp = message.getTs();
- }
Exchange exchange = createExchange(message);
answer.add(exchange);
}
@@ -171,6 +180,7 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
@Override
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
+ int answer = 0;
for (int index = 0; index < total && isBatchAllowed(); index++) {
// only loop if we are started (allowed to run)
@@ -186,9 +196,29 @@ public class SlackConsumer extends
ScheduledBatchPollingConsumer {
// use default consumer callback
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
+ answer++;
+ }
+
+ // the batch ended early as the consumer is suspending or stopping:
the following polls do not read these
+ // messages again, so keep them for the next poll. Their exchanges are
never routed, and nothing else gives
+ // them back to the exchange factory
+ Exchange notRoutedExchange;
+ while ((notRoutedExchange = (Exchange) exchanges.poll()) != null) {
+ notRouted.add(notRoutedExchange.getIn().getBody(Message.class));
+ releaseExchange(notRoutedExchange, true);
+ }
+ if (slackEndpoint.isNaturalOrder()) {
+ // kept the most recent first
+ Collections.reverse(notRouted);
}
+ if (!notRouted.isEmpty()) {
+ LOG.debug("The batch of channel {} ended early, {} messages are
kept for the next poll",
+ slackEndpoint.getChannel(), notRouted.size());
+ }
+ // nothing of this batch is pending anymore: the messages that were
not routed are kept for the next poll
+ pendingExchanges = 0;
- return total;
+ return answer;
}
private String getChannelId(final String channel, final String cursor) {
diff --git
a/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchPooledExchangeTest.java
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchPooledExchangeTest.java
new file mode 100644
index 000000000000..95a016a29fd0
--- /dev/null
+++
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchPooledExchangeTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.slack;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.spi.ExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The same with the pooled exchange factory: the exchanges created for the
messages that a suspending consumer does not
+ * route are given back to the pool, once, so the messages routed after the
resume do not share an exchange.
+ */
+class SlackConsumerSuspendedBatchPooledExchangeTest extends
SlackConsumerSuspendedBatchTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ PooledExchangeFactory factory = new PooledExchangeFactory();
+ factory.setStatisticsEnabled(true);
+ context.getCamelContextExtension().setExchangeFactory(factory);
+ return context;
+ }
+
+ @Test
+ void exchangesOfTheMessagesThatWereNotRoutedAreGivenBackToThePool() throws
Exception {
+
context.getCamelContextExtension().getExchangeFactoryManager().setStatisticsEnabled(true);
+ startRouteAndReceiveFirstMessage();
+ suspendAtMessage = "m4";
+
+ postBurst();
+ awaitRouteSuspended();
+ assertEquals(List.of("m5", "m4"), routed);
+ // one exchange for the first message, and one per message of the
burst: two were routed, and the two that
+ // were not are given back as well
+ awaitExchangesTakenAndReleased(5);
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("m5", "m4", "m3", "m2"), routed,
"messages routed after the resume"));
+ // and only once: the pool does not hold an exchange twice
+ awaitExchangesTakenAndReleased(7);
+ }
+
+ private void awaitExchangesTakenAndReleased(int expected) {
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> {
+ ExchangeFactory.Statistics stats
+ =
context.getCamelContextExtension().getExchangeFactoryManager().getStatistics();
+ assertEquals(expected, stats.getCreatedCounter() +
stats.getAcquiredCounter(), "exchanges taken");
+ assertEquals(expected, stats.getReleasedCounter(), "exchanges
given back to the pool");
+ assertEquals(0, stats.getDiscardedCounter(), "exchanges
discarded");
+ });
+ }
+}
diff --git
a/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchTest.java
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchTest.java
new file mode 100644
index 000000000000..0cea42a77b13
--- /dev/null
+++
b/components/camel-slack/src/test/java/org/apache/camel/component/slack/SlackConsumerSuspendedBatchTest.java
@@ -0,0 +1,311 @@
+/*
+ * 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.slack;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.math.BigDecimal;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+
+import com.slack.api.model.Message;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * A route that is suspended while the consumer routes the messages of a poll,
and resumed: the messages of that poll
+ * that were not routed must be routed after the resume. The consumer has
moved its timestamp past them, so the
+ * following polls do not read them again.
+ */
+class SlackConsumerSuspendedBatchTest extends CamelTestSupport {
+
+ static final String ROUTE_ID = "slack";
+
+ private final List<String> channel = new CopyOnWriteArrayList<>();
+ private HttpServer server;
+ private final AtomicInteger historyRequests = new AtomicInteger();
+ private boolean naturalOrder;
+
+ final List<String> routed = new CopyOnWriteArrayList<>();
+ // the text of the message whose processor suspends the route (once)
+ volatile String suspendAtMessage;
+ // suspend the whole CamelContext instead of the route
+ private volatile boolean suspendContext;
+ private final AtomicBoolean suspendRequested = new AtomicBoolean();
+ private final AtomicReference<Exception> suspendFailure = new
AtomicReference<>();
+ private volatile Thread suspender;
+
+ @Override
+ protected void doPreSetup() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ server.createContext("/api/conversations.list",
this::conversationsList);
+ server.createContext("/api/conversations.history",
this::conversationsHistory);
+ server.start();
+ }
+
+ @AfterEach
+ void stopServer() {
+ // stop polling before the server goes away
+ context.stop();
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void messagesNotRoutedByASuspendingConsumerAreRoutedAfterTheResume()
throws Exception {
+ startRouteAndReceiveFirstMessage();
+ suspendAtMessage = "m4";
+
+ // the most recent message is routed first
+ postBurst();
+ awaitRouteSuspended();
+ assertEquals(List.of("m5", "m4"), routed, "the suspend ends the batch
after the message that is being routed");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("m5", "m4", "m3", "m2"), routed,
"messages routed after the resume"));
+ assertNothingElseIsRouted(List.of("m5", "m4", "m3", "m2"));
+ }
+
+ @Test
+ void
messagesNotRoutedByASuspendingConsumerAreRoutedAfterTheResumeInNaturalOrder()
throws Exception {
+ naturalOrder = true;
+ startRouteAndReceiveFirstMessage();
+ suspendAtMessage = "m3";
+
+ postBurst();
+ awaitRouteSuspended();
+ assertEquals(List.of("m2", "m3"), routed, "the suspend ends the batch
after the message that is being routed");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("m2", "m3", "m4", "m5"), routed,
"messages routed after the resume"));
+ assertNothingElseIsRouted(List.of("m2", "m3", "m4", "m5"));
+ }
+
+ @Test
+ void messagesPostedWhileSuspendedAreRoutedWithTheOnesThatWereNotRouted()
throws Exception {
+ startRouteAndReceiveFirstMessage();
+ suspendAtMessage = "m5";
+
+ postBurst();
+ awaitRouteSuspended();
+ assertEquals(List.of("m5"), routed);
+ channel.add("m6");
+
+ context.getRouteController().resumeRoute(ROUTE_ID);
+ // the most recent first: the new message, then the ones that were left
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("m5", "m6", "m4", "m3", "m2"),
routed, "messages routed after the resume"));
+ assertNothingElseIsRouted(List.of("m5", "m6", "m4", "m3", "m2"));
+ }
+
+ @Test
+ void
messagesNotRoutedWhenTheCamelContextIsSuspendedAreRoutedAfterTheResume() throws
Exception {
+ suspendContext = true;
+ startRouteAndReceiveFirstMessage();
+ suspendAtMessage = "m4";
+
+ postBurst();
+ awaitRouteSuspended();
+ assertEquals(List.of("m5", "m4"), routed, "the suspend ends the batch
after the message that is being routed");
+
+ // the same consumer is resumed
+ context.resume();
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(List.of("m5", "m4", "m3", "m2"), routed,
"messages routed after the resume"));
+ assertNothingElseIsRouted(List.of("m5", "m4", "m3", "m2"));
+ }
+
+ void startRouteAndReceiveFirstMessage() throws Exception {
+ channel.add("m1");
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
fromF("slack://general?token=RAW(xoxp-test)&serverUrl=http://localhost:%d&maxResults=10&delay=20";
+ + "&naturalOrder=%s",
+ server.getAddress().getPort(), naturalOrder)
+ .routeId(ROUTE_ID)
+ .process(exchange -> {
+ String text =
exchange.getMessage().getBody(Message.class).getText();
+ if (text.equals(suspendAtMessage)) {
+ suspendRouteDuringThePoll();
+ }
+ routed.add(text);
+ });
+ }
+ });
+ context.start();
+
+ // the first poll reads the last message of the history
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(List.of("m1"), routed));
+ routed.clear();
+ }
+
+ void postBurst() {
+ // added at once, so that no poll sees only a part of the burst
+ channel.addAll(List.of("m2", "m3", "m4", "m5"));
+ }
+
+ private void assertNothingElseIsRouted(List<String> expected) {
+ // a few more polls: nothing is routed twice
+ int before = historyRequests.get();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> historyRequests.get()
>= before + 3);
+ assertEquals(expected, routed);
+ }
+
+ /**
+ * Called by the thread that polls: suspends the route from another
thread, as a route must not suspend itself, and
+ * returns when the suspend has reached the consumer (it no longer routes
the following messages of the batch).
+ */
+ private void suspendRouteDuringThePoll() {
+ if (!suspendRequested.compareAndSet(false, true)) {
+ return;
+ }
+ SlackConsumer consumer = (SlackConsumer)
context.getRoute(ROUTE_ID).getConsumer();
+ suspender = new Thread(() -> {
+ try {
+ if (suspendContext) {
+ context.suspend();
+ } else {
+ context.getRouteController().suspendRoute(ROUTE_ID);
+ }
+ } catch (Exception e) {
+ suspendFailure.set(e);
+ }
+ }, "suspend-route");
+ suspender.start();
+
+ CountDownLatch batchEnded = new CountDownLatch(1);
+ Thread watcher = new Thread(() -> {
+ await().atMost(20, TimeUnit.SECONDS).until(() ->
!consumer.isBatchAllowed());
+ batchEnded.countDown();
+ }, "watch-consumer");
+ watcher.start();
+
+ // an interrupt of the thread that polls must not end this wait
+ boolean interrupted = false;
+ boolean ended = false;
+ long remaining = TimeUnit.SECONDS.toNanos(30);
+ long deadline = System.nanoTime() + remaining;
+ while (!ended && remaining > 0) {
+ try {
+ ended = batchEnded.await(remaining, TimeUnit.NANOSECONDS);
+ } catch (InterruptedException e) {
+ interrupted = true;
+ }
+ remaining = deadline - System.nanoTime();
+ }
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ void awaitRouteSuspended() throws InterruptedException {
+ await().atMost(20, TimeUnit.SECONDS).until(() -> suspender != null);
+ suspender.join(TimeUnit.SECONDS.toMillis(60));
+ assertFalse(suspender.isAlive(), "the route suspend has returned");
+ assertNull(suspendFailure.get());
+ assertEquals(ServiceStatus.Suspended,
context.getRouteController().getRouteStatus(ROUTE_ID));
+ }
+
+ private void conversationsList(HttpExchange http) throws IOException {
+ respond(http,
"{\"ok\":true,\"channels\":[{\"id\":\"C1\",\"name\":\"general\"}],"
+ + "\"response_metadata\":{\"next_cursor\":\"\"}}");
+ }
+
+ /**
+ * Answers as Slack does: the messages after {@code oldest} (exclusive),
the most recent first, {@code limit} at a
+ * time.
+ */
+ private void conversationsHistory(HttpExchange http) throws IOException {
+ historyRequests.incrementAndGet();
+ Map<String, String> params = form(http);
+ String oldest = params.get("oldest");
+ int limit = Integer.parseInt(params.get("limit"));
+
+ List<Integer> range = new ArrayList<>();
+ for (int i = 0; i < channel.size(); i++) {
+ if (oldest == null || new BigDecimal(ts(i)).compareTo(new
BigDecimal(oldest)) > 0) {
+ range.add(i);
+ }
+ }
+ range.sort(Comparator.reverseOrder());
+ List<Integer> page = range.subList(0, Math.min(limit, range.size()));
+
+ String messages = page.stream()
+ .map(i -> "{\"type\":\"message\",\"text\":\"" + channel.get(i)
+ "\",\"ts\":\"" + ts(i) + "\"}")
+ .collect(Collectors.joining(","));
+ respond(http, "{\"ok\":true,\"messages\":[" + messages +
"],\"has_more\":false,"
+ + "\"response_metadata\":{\"next_cursor\":\"\"}}");
+ }
+
+ private static String ts(int index) {
+ return (1700000000 + index) + ".000100";
+ }
+
+ private static Map<String, String> form(HttpExchange http) throws
IOException {
+ Map<String, String> answer = new HashMap<>();
+ String body = new String(http.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ for (String pair : body.split("&")) {
+ int idx = pair.indexOf('=');
+ if (idx > 0) {
+ answer.put(URLDecoder.decode(pair.substring(0, idx),
StandardCharsets.UTF_8),
+ URLDecoder.decode(pair.substring(idx + 1),
StandardCharsets.UTF_8));
+ }
+ }
+ return answer;
+ }
+
+ private static void respond(HttpExchange http, String json) throws
IOException {
+ byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
+ http.getResponseHeaders().add("Content-Type", "application/json;
charset=utf-8");
+ http.sendResponseHeaders(200, bytes.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 8a6f106f08a6..8bcccd225fc9 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -932,6 +932,14 @@ failed (for example a 404 for a deleted account, or a
server error). Such a comm
for it. Its `CamelGitHubCommitAuthor` and `CamelGitHubCommitCommitter` headers
then have the name that git recorded in
the commit instead of the name of the GitHub profile, and are not set when the
commit has no name either.
+=== camel-slack - consumer and a suspended route
+
+When a route is suspended (`RouteController.suspendRoute`, also from a route
policy that suspends the route such as
+`DurationRoutePolicy`, or `CamelContext.suspend()`) while the `slack` consumer
routes the messages of a poll, the
+consumer stops after the message that is being routed. The other messages of
that poll were skipped when the route
+was resumed. They are now kept and routed by the first poll after the resume,
together with the messages posted in
+the meantime. They are kept in memory only: a route that is stopped and
started again does not route them.
+
=== camel-smpp - the host name of the SMSC is verified over SSL
With `smpps` (or `usingSSL=true`), directly or through an HTTP proxy, the
connection now checks that the certificate of
