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 f245636df70f CAMEL-25712: camel-google-pubsub - keep pulling after a
failed nack of a message at max delivery attempts (#27992)
f245636df70f is described below
commit f245636df70f57114750d30b1270bb9dc18f30af
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 14:42:42 2026 +0530
CAMEL-25712: camel-google-pubsub - keep pulling after a failed nack of a
message at max delivery attempts (#27992)
With synchronousPull=true and maxDeliveryAttempts in effect (set on the
endpoint or read from the dead-letter policy of the subscription), a
message that reached the limit is nacked without routing. The nack is a
ModifyAckDeadline call; AcknowledgeSync wraps its failure in a
RuntimeCamelException that no catch of the pull loop handles, so the
exception ended the consumer thread. The consumer stayed started and
never pulled again; only an ERROR "Failure getting messages from PubSub"
was logged.
The failure is now reported through the consumer's ExceptionHandler and
the loop goes on. The message is delivered again when its ack deadline
expires and is nacked then.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../google/pubsub/GooglePubsubConsumer.java | 11 +-
.../GooglePubsubConsumerNackFailureTest.java | 130 +++++++++++++++++++++
2 files changed, 140 insertions(+), 1 deletion(-)
diff --git
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
index ac94604b3cec..c6198be41605 100644
---
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
+++
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumer.java
@@ -362,7 +362,16 @@ public class GooglePubsubConsumer extends DefaultConsumer
implements ShutdownAwa
pubsubMessage.getMessageId(),
deliveryAttempt, resolvedMaxDeliveryAttempts);
GooglePubsubAcknowledge ack = new AcknowledgeSync(
() ->
endpoint.getComponent().getSubscriberStub(endpoint), subscriptionName);
- ack.nack(exchange);
+ try {
+ ack.nack(exchange);
+ } catch (Exception e) {
+ // keep pulling: the message is delivered
again when its ack deadline expires and is
+ // nacked then, while an exception here would
end the pull loop of this consumer
+ getExceptionHandler().handleException(
+ "Error nacking message " +
pubsubMessage.getMessageId()
+ + " that
reached max delivery attempts",
+ exchange, e);
+ }
continue;
}
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerNackFailureTest.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerNackFailureTest.java
new file mode 100644
index 000000000000..6b4b75b4482f
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerNackFailureTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.google.pubsub;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.UnaryCallable;
+import com.google.api.gax.rpc.UnavailableException;
+import com.google.cloud.pubsub.v1.stub.SubscriberStub;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Empty;
+import com.google.pubsub.v1.AcknowledgeRequest;
+import com.google.pubsub.v1.ModifyAckDeadlineRequest;
+import com.google.pubsub.v1.PubsubMessage;
+import com.google.pubsub.v1.PullRequest;
+import com.google.pubsub.v1.PullResponse;
+import com.google.pubsub.v1.ReceivedMessage;
+import io.grpc.Status;
+import org.apache.camel.CamelContext;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * With synchronous pull, a failure to nack a message that reached
maxDeliveryAttempts must not end the pull loop: the
+ * consumer must keep receiving the next messages.
+ */
+public class GooglePubsubConsumerNackFailureTest {
+
+ @Test
+ void failedNackOfMessageAtMaxDeliveryAttemptsDoesNotStopPolling() throws
Exception {
+ AtomicInteger pulls = new AtomicInteger();
+ SettableApiFuture<PullResponse> pending = SettableApiFuture.create();
+
+ @SuppressWarnings("unchecked")
+ UnaryCallable<PullRequest, PullResponse> pullCallable =
mock(UnaryCallable.class);
+
when(pullCallable.futureCall(any(PullRequest.class))).thenAnswer(invocation -> {
+ int pull = pulls.incrementAndGet();
+ if (pull == 1) {
+ // a poison message that reached the max delivery attempts:
nacked without routing
+ return ApiFutures.immediateFuture(response("ack-1", "poison",
3));
+ } else if (pull == 2) {
+ return ApiFutures.immediateFuture(response("ack-2", "hello",
1));
+ }
+ // no more messages; cancelled when the consumer stops
+ return pending;
+ });
+
+ @SuppressWarnings("unchecked")
+ UnaryCallable<ModifyAckDeadlineRequest, Empty>
modifyAckDeadlineCallable = mock(UnaryCallable.class);
+
when(modifyAckDeadlineCallable.call(any(ModifyAckDeadlineRequest.class)))
+ .thenThrow(new UnavailableException(
+ "Pub/Sub unavailable", null,
GrpcStatusCode.of(Status.Code.UNAVAILABLE), true));
+
+ @SuppressWarnings("unchecked")
+ UnaryCallable<AcknowledgeRequest, Empty> acknowledgeCallable =
mock(UnaryCallable.class);
+
when(acknowledgeCallable.call(any(AcknowledgeRequest.class))).thenReturn(Empty.getDefaultInstance());
+
+ SubscriberStub stub = mock(SubscriberStub.class);
+ when(stub.pullCallable()).thenReturn(pullCallable);
+
when(stub.modifyAckDeadlineCallable()).thenReturn(modifyAckDeadlineCallable);
+ when(stub.acknowledgeCallable()).thenReturn(acknowledgeCallable);
+
+ GooglePubsubComponent component = new GooglePubsubComponent() {
+ @Override
+ public SubscriberStub getSubscriberStub(GooglePubsubEndpoint
googlePubsubEndpoint) {
+ return stub;
+ }
+ };
+
+ CamelContext context = new DefaultCamelContext();
+ try {
+ context.addComponent("google-pubsub", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("google-pubsub:test-project:test-subscription?synchronousPull=true&maxDeliveryAttempts=3")
+ .routeId("pubsub")
+ .to("mock:result");
+ }
+ });
+ MockEndpoint result = context.getEndpoint("mock:result",
MockEndpoint.class);
+ result.expectedBodiesReceived("hello");
+ result.setResultWaitTime(10000);
+
+ context.start();
+
+ result.assertIsSatisfied();
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus("pubsub"));
+ } finally {
+ pending.cancel(true);
+ context.stop();
+ }
+ }
+
+ private static PullResponse response(String ackId, String body, int
deliveryAttempt) {
+ return PullResponse.newBuilder()
+ .addReceivedMessages(ReceivedMessage.newBuilder()
+ .setAckId(ackId)
+ .setDeliveryAttempt(deliveryAttempt)
+ .setMessage(PubsubMessage.newBuilder()
+ .setMessageId(ackId)
+ .setData(ByteString.copyFromUtf8(body))))
+ .build();
+ }
+}
