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 914328b3555f CAMEL-25577: camel-telegram - Complete the exchange once
when Telegram answers with an error (#27761)
914328b3555f is described below
commit 914328b3555f35ccec2b2f2c984091d1fb31fbc9
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 13 12:26:01 2026 +0530
CAMEL-25577: camel-telegram - Complete the exchange once when Telegram
answers with an error (#27761)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../telegram/service/TelegramAsyncHandler.java | 12 +-
.../telegram/service/TelegramMessageHandler.java | 23 +-
.../telegram/TelegramProducerFailureTest.java | 303 +++++++++++++++++++++
3 files changed, 324 insertions(+), 14 deletions(-)
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramAsyncHandler.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramAsyncHandler.java
index 3b180356d1a8..50b3f0bb18d3 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramAsyncHandler.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramAsyncHandler.java
@@ -26,7 +26,6 @@ import java.net.http.HttpResponse;
import java.nio.charset.StandardCharsets;
import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.telegram.model.MessageResult;
@@ -42,15 +41,12 @@ public class TelegramAsyncHandler {
private final Class<? extends MessageResult> resultClass;
private final ObjectMapper mapper;
private final Exchange exchange;
- private final AsyncCallback callback;
- TelegramAsyncHandler(String uri, Class<? extends MessageResult>
resultClass, ObjectMapper mapper, Exchange exchange,
- AsyncCallback callback) {
+ TelegramAsyncHandler(String uri, Class<? extends MessageResult>
resultClass, ObjectMapper mapper, Exchange exchange) {
this.uri = uri;
this.resultClass = resultClass;
this.mapper = mapper;
this.exchange = exchange;
- this.callback = callback;
}
public static String extractCharset(String line, String defaultValue) {
@@ -78,6 +74,10 @@ public class TelegramAsyncHandler {
}
+ /**
+ * Sets the result of a successful response as the message body. Throws
for a response with an error status and for
+ * a body that cannot be read; the caller fails the exchange with that
exception and completes it.
+ */
public Object handleCompressedResponse(HttpResponse<InputStream> response)
{
Object result;
final boolean success = response.statusCode() >= 200 &&
response.statusCode() < 300;
@@ -112,8 +112,6 @@ public class TelegramAsyncHandler {
throw new RuntimeException(ex);
} catch (IOException ex) {
throw new RuntimeCamelException("Could not parse the response from
" + uri, ex);
- } finally {
- callback.done(false);
}
return exchange;
diff --git
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
index a2f9738ac25e..d92da68044a0 100644
---
a/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
+++
b/components/camel-telegram/src/main/java/org/apache/camel/component/telegram/service/TelegramMessageHandler.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.telegram.service;
import java.net.URI;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -63,18 +64,26 @@ abstract class TelegramMessageHandler<T extends
TelegramMessage> {
builder.setHeader("Accept", "application/json");
builder.POST(bodyPublisher.newPublisher());
- try {
- final TelegramAsyncHandler telegramAsyncHandler
- = new TelegramAsyncHandler(uri, resultClass, mapper,
exchange, callback);
+ final TelegramAsyncHandler telegramAsyncHandler = new
TelegramAsyncHandler(uri, resultClass, mapper, exchange);
+ final CompletableFuture<Object> response
+ = apiClient.client().sendAsync(builder.build(),
HttpResponse.BodyHandlers.ofInputStream())
+
.thenApply(telegramAsyncHandler::handleCompressedResponse);
- apiClient.client().sendAsync(builder.build(),
HttpResponse.BodyHandlers.ofInputStream())
-
.thenApply(telegramAsyncHandler::handleCompressedResponse).get();
+ try {
+ // the calling thread waits for the answer, so a caller sends one
message after the other
+ response.get();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
- throw new RuntimeException(e);
+ // nobody waits for the answer any more: keep the stage from
writing to the completed exchange later
+ response.cancel(true);
+ exchange.setException(new RuntimeException(e));
} catch (ExecutionException e) {
- throw new RuntimeException(e);
+ // an error status, a body that could not be read, or a request
that failed without an answer
+ exchange.setException(new RuntimeException(e));
}
+
+ // the only place that completes the exchange. The producer returns
false from process(), hence done(false)
+ callback.done(false);
}
protected abstract void addBody(T message);
diff --git
a/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerFailureTest.java
b/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerFailureTest.java
new file mode 100644
index 000000000000..b90c81755157
--- /dev/null
+++
b/components/camel-telegram/src/test/java/org/apache/camel/component/telegram/TelegramProducerFailureTest.java
@@ -0,0 +1,303 @@
+/*
+ * 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.telegram;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.AsyncProducer;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.telegram.model.MessageResult;
+import org.apache.camel.component.telegram.model.OutgoingTextMessage;
+import org.apache.camel.component.telegram.model.SendLocationMessage;
+import org.apache.camel.component.telegram.model.SendVenueMessage;
+import org.apache.camel.component.telegram.util.TelegramMockRoutes;
+import org.apache.camel.component.telegram.util.TelegramTestSupport;
+import org.apache.camel.component.telegram.util.TelegramTestUtil;
+import org.apache.camel.test.AvailablePortFinder;
+import org.assertj.core.api.SoftAssertions;
+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.SoftAssertions.assertSoftly;
+
+/**
+ * A request that Telegram answers with an error, with a body that cannot be
read, or not at all, must fail the exchange
+ * and complete it once, and the calling thread waits for the answer.
+ */
+class TelegramProducerFailureTest extends TelegramTestSupport {
+
+ // reserved for this test, and nothing listens on it
+ @RegisterExtension
+ private static final AvailablePortFinder.Port CLOSED_PORT =
AvailablePortFinder.find();
+
+ private final HoldingApi venueApi = new HoldingApi();
+
+ @EndpointInject("mock:after")
+ private MockEndpoint after;
+
+ @EndpointInject("mock:failed")
+ private MockEndpoint failed;
+
+ @Test
+ void errorStatusFailsTheExchangeOnce() throws Exception {
+ Throwable failure = assertFailsOnce("direct:telegram", telegramUri(),
textMessage());
+
+
assertThat(failure).isInstanceOf(RuntimeCamelException.class).hasMessageContaining("responded:
400");
+ }
+
+ @Test
+ void unreadableAnswerFailsTheExchangeOnce() throws Exception {
+ Throwable failure = assertFailsOnce("direct:telegram", telegramUri(),
new SendLocationMessage(59.9386292, 30.3141308));
+
+
assertThat(failure).isInstanceOf(RuntimeCamelException.class).hasMessageContaining("Could
not parse the response");
+ }
+
+ @Test
+ void requestWithoutAnswerFailsTheExchangeOnce() throws Exception {
+ Throwable failure = assertFailsOnce("direct:closed", closedPortUri(),
textMessage());
+
+ assertThat(failure).isInstanceOf(IOException.class);
+ }
+
+ @Test
+ void errorStatusIsHandledByOnExceptionAndTheRouteDoesNotContinue() {
+ Exchange result = template.send("direct:handled", e ->
e.getIn().setBody(textMessage()));
+
+ assertSoftly(softly -> {
+ softly.assertThat(after.getReceivedCounter()).as("exchanges at the
step after the producer").isZero();
+ softly.assertThat(failed.getReceivedCounter()).as("exchanges at
onException").isEqualTo(1);
+ softly.assertThat(result.getException()).as("exception left on the
handled exchange").isNull();
+ });
+ }
+
+ @Test
+ void interruptedCallerFailsTheExchangeOnce() throws Exception {
+ Completion completion;
+ boolean stillInterrupted;
+ Thread.currentThread().interrupt();
+ try {
+ completion = process(telegramUri(), textMessage());
+ } finally {
+ // also clears the flag for the rest of the test
+ stillInterrupted = Thread.interrupted();
+ }
+
+ assertSoftly(softly -> completedOnce(softly, completion, true));
+ assertThat(stillInterrupted).as("interrupt flag kept").isTrue();
+
assertThat(completion.exchange.getException()).isExactlyInstanceOf(RuntimeException.class)
+ .hasCauseInstanceOf(InterruptedException.class);
+ }
+
+ /**
+ * The order of the requests is the same without the fix: the calling
thread has always waited for the answer, and
+ * it must stay so. What is new for an answered message is the thread that
completes the exchange.
+ */
+ @Test
+ void twoSendsAreNotInterleavedAndSetTheResult() throws Exception {
+ AsyncProducer producer =
context.getEndpoint(telegramUri()).createAsyncProducer();
+ producer.start();
+ try {
+ for (int i = 0; i < 2; i++) {
+ Completion completion = process(producer, new
SendVenueMessage(59.9386292, 30.3141308, "title", "address"));
+
+ assertSoftly(softly -> completedOnce(softly, completion,
false));
+ assertThat(completion.exchange.getException()).isNull();
+
assertThat(completion.exchange.getMessage().getBody()).isInstanceOf(MessageResult.class);
+ }
+ } finally {
+ producer.stop();
+ }
+
+ assertThat(venueApi.events).containsExactly("request 1", "answer 1",
"request 2", "answer 2");
+ }
+
+ /**
+ * Sends the message through a route and with the producer alone, and
returns the failure that Telegram caused.
+ */
+ private Throwable assertFailsOnce(String routeUri, String producerUri,
Object body) throws Exception {
+ Exchange routed = template.send(routeUri, e ->
e.getIn().setBody(body));
+ Completion completion = process(producerUri, body);
+
+ assertSoftly(softly -> {
+ softly.assertThat(after.getReceivedCounter()).as("exchanges at the
step after the producer").isZero();
+ softly.assertThat(routed.getException()).as("exception of the
routed exchange").isNotNull();
+ completedOnce(softly, completion, true);
+ });
+ // the same exception as before, when process() threw it
+ Throwable exception = completion.exchange.getException();
+
assertThat(exception).isExactlyInstanceOf(RuntimeException.class).hasCauseInstanceOf(ExecutionException.class);
+
assertThat(routed.getException()).isExactlyInstanceOf(RuntimeException.class)
+ .hasCauseInstanceOf(ExecutionException.class);
+ return exception.getCause().getCause();
+ }
+
+ private static void completedOnce(SoftAssertions softly, Completion
completion, boolean failed) {
+ softly.assertThat(completion.thrownByProcess).as("thrown by process()
of the producer").isNull();
+ softly.assertThat(completion.callsWhenProcessReturned).as("calls of
the callback when process() returned")
+ .isEqualTo(1);
+ softly.assertThat(completion.calls.get()).as("calls of the
callback").isEqualTo(1);
+ softly.assertThat(completion.callingThread.get()).as("thread that
called the callback")
+ .isSameAs(Thread.currentThread());
+ softly.assertThat(completion.failedAtFirstCall.get()).as("exchange
failed when it was completed")
+ .isEqualTo(failed);
+ }
+
+ private Completion process(String uri, Object body) throws Exception {
+ AsyncProducer producer =
context.getEndpoint(uri).createAsyncProducer();
+ producer.start();
+ try {
+ return process(producer, body);
+ } finally {
+ producer.stop();
+ }
+ }
+
+ /**
+ * Calls the producer the way a route does and records every call of the
callback.
+ */
+ private static Completion process(AsyncProducer producer, Object body)
throws Exception {
+ Exchange exchange = producer.getEndpoint().createExchange();
+ exchange.getIn().setBody(body);
+ Completion completion = new Completion(exchange);
+ try {
+ producer.process(exchange, doneSync -> {
+ if (completion.calls.getAndIncrement() == 0) {
+ completion.callingThread.set(Thread.currentThread());
+ completion.failedAtFirstCall.set(exchange.getException()
!= null);
+ }
+ completion.called.countDown();
+ });
+ } catch (RuntimeException e) {
+ completion.thrownByProcess = e;
+ }
+ completion.callsWhenProcessReturned = completion.calls.get();
+ if (completion.thrownByProcess == null &&
completion.callsWhenProcessReturned == 0) {
+ // a producer that does not wait for the answer gets the time to
complete the exchange, so nothing hangs
+ assertThat(completion.called.await(20, TimeUnit.SECONDS))
+ .as("the exchange was completed within 20
seconds").isTrue();
+ }
+ return completion;
+ }
+
+ private String telegramUri() {
+ return "telegram:bots?authorizationToken=mock-token&chatId=" + chatId;
+ }
+
+ private String closedPortUri() {
+ return telegramUri() + "&baseUri=http://localhost:"; +
CLOSED_PORT.getPort();
+ }
+
+ private static OutgoingTextMessage textMessage() {
+ OutgoingTextMessage message = new OutgoingTextMessage();
+ message.setText("hello");
+ return message;
+ }
+
+ private static final class Completion {
+ private final Exchange exchange;
+ private final AtomicInteger calls = new AtomicInteger();
+ private final AtomicReference<Boolean> failedAtFirstCall = new
AtomicReference<>();
+ private final AtomicReference<Thread> callingThread = new
AtomicReference<>();
+ private final CountDownLatch called = new CountDownLatch(1);
+ // written and read by the test thread only
+ private RuntimeException thrownByProcess;
+ private int callsWhenProcessReturned;
+
+ private Completion(Exchange exchange) {
+ this.exchange = exchange;
+ }
+ }
+
+ /**
+ * Answers sendVenue. It keeps the first answer back for a second, or
until a second request arrives: a producer
+ * that sends the second message before the first is answered shows up in
the order of the events.
+ */
+ private static final class HoldingApi implements Processor {
+ private final List<String> events = new CopyOnWriteArrayList<>();
+ private final AtomicInteger requests = new AtomicInteger();
+ private final CountDownLatch secondArrived = new CountDownLatch(1);
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ int request = requests.incrementAndGet();
+ events.add("request " + request);
+ if (request == 1) {
+ if (secondArrived.await(1, TimeUnit.SECONDS)) {
+ events.add("request 2 arrived before answer 1");
+ }
+ } else {
+ secondArrived.countDown();
+ }
+ events.add("answer " + request);
+ byte[] bytes =
TelegramTestUtil.stringResource("messages/send-venue.json").getBytes(StandardCharsets.UTF_8);
+ exchange.getMessage().setBody(bytes);
+ exchange.getMessage().setHeader("Content-Length", bytes.length);
+ exchange.getMessage().setHeader("Content-Type", "application/json;
charset=UTF-8");
+ }
+ }
+
+ @Override
+ protected RoutesBuilder[] createRouteBuilders() {
+ return new RoutesBuilder[] {
+ getMockRoutes(),
+ new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("netty-http:http://localhost:"; + port.getPort()
+ +
"/botmock-token/sendVenue?httpMethodRestrict=POST")
+ .process(venueApi);
+
+ from("direct:telegram")
+ .to(telegramUri())
+ .to("mock:after");
+
+ from("direct:closed")
+ .to(closedPortUri())
+ .to("mock:after");
+
+ from("direct:handled")
+
.onException(Exception.class).handled(true).to("mock:failed").end()
+ .to(telegramUri())
+ .to("mock:after");
+ }
+ } };
+ }
+
+ @Override
+ protected TelegramMockRoutes createMockRoutes() {
+ return new TelegramMockRoutes(port.getPort())
+ // Telegram answers 400, for example for a chat that does not
exist
+ .addErrorEndpoint("sendMessage", "POST", 400)
+ // 200 with a body that is not JSON, for example the error
page of a proxy
+ .addEndpoint("sendLocation", "POST", String.class, "<html>Bad
Gateway</html>");
+ }
+}
