This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 73aa5da9c0d9 [camel-4.22.x] CAMEL-26647: camel-google-pubsub - a
manual ack of an aggregated exchange acks every grouped message with
synchronousPull (#29350)
73aa5da9c0d9 is described below
commit 73aa5da9c0d93b35c4a1faa3e988e528c141d154
Author: Federico Mariani <[email protected]>
AuthorDate: Fri Nov 6 21:38:34 2026 +0100
[camel-4.22.x] CAMEL-26647: camel-google-pubsub - a manual ack of an
aggregated exchange acks every grouped message with synchronousPull (#29350)
Co-authored-by: Claude <[email protected]>
---
.../catalog/docs/google-pubsub-component.adoc | 26 +-
.../src/main/docs/google-pubsub-component.adoc | 26 +-
.../google/pubsub/consumer/AcknowledgeSync.java | 50 +++-
.../GooglePubsubConsumerAggregatedAckTest.java | 263 +++++++++++++++++++++
.../component/google/pubsub/PubsubTestSupport.java | 17 +-
.../AggregatedManualAcknowledgementIT.java | 109 +++++++++
6 files changed, 466 insertions(+), 25 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/google-pubsub-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/google-pubsub-component.adoc
index d1c93f3d4aff..2041b152c237 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/google-pubsub-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/google-pubsub-component.adoc
@@ -84,8 +84,8 @@ If the route throws an exception, the exchange is marked as
failed, and the comp
It will be redelivered immediately.
To ack/nack the message the component uses Acknowledgement ID stored as header
`GooglePubsubConstants.ACK_ID`.
-If the header is removed or tampered with, the ack will fail and the message
will be redelivered
-again after the ack deadline.
+With the synchronous consumer (`synchronousPull=true`), if the header is
changed, the ack fails; if it is removed,
+no ack is sent and a WARN is logged. In both cases the message is redelivered
after the ack deadline.
=== Delivery Attempts and Dead Letter Queues
@@ -231,4 +231,24 @@
from("google-pubsub:{{project.name}}:{{subscription.name}}?ackMode=NONE")
});
----
-Manual acknowledgement works with both the asynchronous and synchronous
consumers and will use the acknowledgement id which is stored in
`GooglePubsubConstants.ACK_ID`.
+Manual acknowledgement works with both the asynchronous and synchronous
consumers.
+The synchronous consumer (`synchronousPull=true`) uses the acknowledgement id
stored in the header `GooglePubsubConstants.ACK_ID` of the exchange given to
`ack` or `nack`.
+The asynchronous consumer acknowledges the message the
`GooglePubsubAcknowledge` object was taken from, whatever the exchange given.
+
+With the synchronous consumer, `ack` or `nack` of an aggregated exchange whose
body is a `List<Exchange>`, such as the result of the
`GroupedExchangeAggregationStrategy`, applies to every grouped message, in one
request:
+
+._Java-only: manual acknowledgement of an aggregated exchange with the
synchronous consumer_
+[source,java]
+----
+from("google-pubsub:{{project.name}}:{{subscription.name}}?synchronousPull=true&ackMode=NONE&maxMessagesPerPoll=10")
+ .aggregate(constant(true), new
GroupedExchangeAggregationStrategy()).completionSize(10)
+ .process(exchange -> {
+ List<Exchange> grouped = exchange.getIn().getBody(List.class);
+ GooglePubsubAcknowledge acknowledge =
grouped.get(0).getIn().getHeader(GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE,
GooglePubsubAcknowledge.class);
+ acknowledge.ack(exchange); // acknowledges the 10 grouped messages
+ });
+----
+
+The request goes to the subscription of the `GooglePubsubAcknowledge` used, so
every grouped exchange must come from that same subscription, that is from the
same consumer endpoint.
+The synchronous consumer does not extend the ack deadline of the messages it
pulls, so the aggregated exchange must be acknowledged before the ack deadline
of the subscription expires, otherwise the grouped messages are delivered again.
+With the asynchronous consumer, acknowledge each grouped exchange with its own
`GooglePubsubAcknowledge` header.
diff --git
a/components/camel-google/camel-google-pubsub/src/main/docs/google-pubsub-component.adoc
b/components/camel-google/camel-google-pubsub/src/main/docs/google-pubsub-component.adoc
index d1c93f3d4aff..2041b152c237 100644
---
a/components/camel-google/camel-google-pubsub/src/main/docs/google-pubsub-component.adoc
+++
b/components/camel-google/camel-google-pubsub/src/main/docs/google-pubsub-component.adoc
@@ -84,8 +84,8 @@ If the route throws an exception, the exchange is marked as
failed, and the comp
It will be redelivered immediately.
To ack/nack the message the component uses Acknowledgement ID stored as header
`GooglePubsubConstants.ACK_ID`.
-If the header is removed or tampered with, the ack will fail and the message
will be redelivered
-again after the ack deadline.
+With the synchronous consumer (`synchronousPull=true`), if the header is
changed, the ack fails; if it is removed,
+no ack is sent and a WARN is logged. In both cases the message is redelivered
after the ack deadline.
=== Delivery Attempts and Dead Letter Queues
@@ -231,4 +231,24 @@
from("google-pubsub:{{project.name}}:{{subscription.name}}?ackMode=NONE")
});
----
-Manual acknowledgement works with both the asynchronous and synchronous
consumers and will use the acknowledgement id which is stored in
`GooglePubsubConstants.ACK_ID`.
+Manual acknowledgement works with both the asynchronous and synchronous
consumers.
+The synchronous consumer (`synchronousPull=true`) uses the acknowledgement id
stored in the header `GooglePubsubConstants.ACK_ID` of the exchange given to
`ack` or `nack`.
+The asynchronous consumer acknowledges the message the
`GooglePubsubAcknowledge` object was taken from, whatever the exchange given.
+
+With the synchronous consumer, `ack` or `nack` of an aggregated exchange whose
body is a `List<Exchange>`, such as the result of the
`GroupedExchangeAggregationStrategy`, applies to every grouped message, in one
request:
+
+._Java-only: manual acknowledgement of an aggregated exchange with the
synchronous consumer_
+[source,java]
+----
+from("google-pubsub:{{project.name}}:{{subscription.name}}?synchronousPull=true&ackMode=NONE&maxMessagesPerPoll=10")
+ .aggregate(constant(true), new
GroupedExchangeAggregationStrategy()).completionSize(10)
+ .process(exchange -> {
+ List<Exchange> grouped = exchange.getIn().getBody(List.class);
+ GooglePubsubAcknowledge acknowledge =
grouped.get(0).getIn().getHeader(GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE,
GooglePubsubAcknowledge.class);
+ acknowledge.ack(exchange); // acknowledges the 10 grouped messages
+ });
+----
+
+The request goes to the subscription of the `GooglePubsubAcknowledge` used, so
every grouped exchange must come from that same subscription, that is from the
same consumer endpoint.
+The synchronous consumer does not extend the ack deadline of the messages it
pulls, so the aggregated exchange must be acknowledged before the ack deadline
of the subscription expires, otherwise the grouped messages are delivered again.
+With the asynchronous consumer, acknowledge each grouped exchange with its own
`GooglePubsubAcknowledge` header.
diff --git
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/AcknowledgeSync.java
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/AcknowledgeSync.java
index 5cefe15919fb..0ac6f8456a7e 100644
---
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/AcknowledgeSync.java
+++
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/consumer/AcknowledgeSync.java
@@ -17,7 +17,9 @@
package org.apache.camel.component.google.pubsub.consumer;
import java.util.ArrayList;
+import java.util.LinkedHashSet;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.Callable;
import com.google.cloud.pubsub.v1.stub.SubscriberStub;
@@ -26,9 +28,13 @@ import com.google.pubsub.v1.ModifyAckDeadlineRequest;
import org.apache.camel.Exchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.google.pubsub.GooglePubsubConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class AcknowledgeSync implements GooglePubsubAcknowledge {
+ private static final Logger LOG =
LoggerFactory.getLogger(AcknowledgeSync.class);
+
//Supplier cannot be used because of thrown exception (Callback used
instead)
private final Callable<SubscriberStub> subscriberStubSupplier;
private final String subscriptionName;
@@ -40,8 +46,16 @@ public class AcknowledgeSync implements
GooglePubsubAcknowledge {
@Override
public void ack(Exchange exchange) {
+ List<String> ackIds = getAckIdList(exchange);
+ if (ackIds.isEmpty()) {
+ // the API requires at least one ack id; this may run in an on
completion callback, so do not throw
+ LOG.warn("No {} header on exchange {} nor on its grouped
exchanges: no message is acknowledged,"
+ + " so Pub/Sub delivers the messages again after their
ack deadline",
+ GooglePubsubConstants.ACK_ID, exchange.getExchangeId());
+ return;
+ }
AcknowledgeRequest ackRequest = AcknowledgeRequest.newBuilder()
- .addAllAckIds(getAckIdList(exchange))
+ .addAllAckIds(ackIds)
.setSubscription(subscriptionName).build();
try (SubscriberStub subscriber = subscriberStubSupplier.call()) {
subscriber.acknowledgeCallable().call(ackRequest);
@@ -54,8 +68,15 @@ public class AcknowledgeSync implements
GooglePubsubAcknowledge {
public void nack(Exchange exchange) {
// There is no explicit nack on the subscriber client. Using
modifyAckDeadline with 0 seconds
// is the recommended way to nack a message.
https://github.com/googleapis/python-pubsub/pull/123
+ List<String> ackIds = getAckIdList(exchange);
+ if (ackIds.isEmpty()) {
+ LOG.warn("No {} header on exchange {} nor on its grouped
exchanges: no message is negatively acknowledged,"
+ + " so Pub/Sub delivers the messages again after their
ack deadline",
+ GooglePubsubConstants.ACK_ID, exchange.getExchangeId());
+ return;
+ }
ModifyAckDeadlineRequest nackRequest =
ModifyAckDeadlineRequest.newBuilder()
- .addAllAckIds(getAckIdList(exchange))
+ .addAllAckIds(ackIds)
.setSubscription(subscriptionName)
.setAckDeadlineSeconds(0).build();
@@ -67,24 +88,27 @@ public class AcknowledgeSync implements
GooglePubsubAcknowledge {
}
private List<String> getAckIdList(Exchange exchange) {
- List<String> ackList = new ArrayList<>();
+ // an aggregated exchange (e.g. from
GroupedExchangeAggregationStrategy) carries the grouped exchanges as its
+ // body: each one holds the ack id of its own message, and the
aggregated exchange may be one of them
+ Set<String> ackIds = new LinkedHashSet<>();
- if (exchange.getIn().getBody() instanceof List) {
- for (Object body : exchange.getIn().getBody(List.class)) {
- if (body instanceof Exchange) {
- String ackId =
exchange.getIn().getHeader(GooglePubsubConstants.ACK_ID, String.class);
- if (null != ackId) {
- ackList.add(ackId);
- }
+ if (exchange.getIn().getBody() instanceof List<?> grouped) {
+ for (Object body : grouped) {
+ if (body instanceof Exchange groupedExchange) {
+ addAckId(ackIds, groupedExchange);
}
}
}
+ addAckId(ackIds, exchange);
+
+ return new ArrayList<>(ackIds);
+ }
+
+ private static void addAckId(Set<String> ackIds, Exchange exchange) {
String ackId =
exchange.getIn().getHeader(GooglePubsubConstants.ACK_ID, String.class);
if (null != ackId) {
- ackList.add(ackId);
+ ackIds.add(ackId);
}
-
- return ackList;
}
}
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerAggregatedAckTest.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerAggregatedAckTest.java
new file mode 100644
index 000000000000..257586cd973b
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/GooglePubsubConsumerAggregatedAckTest.java
@@ -0,0 +1,263 @@
+/*
+ * 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.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
+
+import com.google.api.core.ApiFutures;
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.rpc.UnaryCallable;
+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 org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.google.pubsub.consumer.AcknowledgeSync;
+import
org.apache.camel.component.google.pubsub.consumer.GooglePubsubAcknowledge;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * With synchronous pull and ackMode=NONE, acknowledging an exchange
aggregated from the consumed ones (a
+ * {@code List<Exchange>} body) must acknowledge every grouped message, each
with its own ack id.
+ */
+public class GooglePubsubConsumerAggregatedAckTest {
+
+ private static final String SUBSCRIPTION =
"projects/test-project/subscriptions/test-subscription";
+
+ private final List<AcknowledgeRequest> ackRequests = new
CopyOnWriteArrayList<>();
+ private final List<ModifyAckDeadlineRequest> modifyAckDeadlineRequests =
new CopyOnWriteArrayList<>();
+ private final SettableApiFuture<PullResponse> pending =
SettableApiFuture.create();
+ private final AtomicInteger stubs = new AtomicInteger();
+
+ @Test
+ void ackOfAggregatedExchangeAcksEveryGroupedMessage() throws Exception {
+ routeAggregatedExchange(GooglePubsubAcknowledge::ack);
+
+ assertEquals(1, ackRequests.size());
+ assertEquals(List.of("ack-1", "ack-2", "ack-3"),
ackRequests.get(0).getAckIdsList());
+ assertEquals(SUBSCRIPTION, ackRequests.get(0).getSubscription());
+ assertTrue(modifyAckDeadlineRequests.isEmpty());
+ }
+
+ @Test
+ void nackOfAggregatedExchangeNacksEveryGroupedMessage() throws Exception {
+ routeAggregatedExchange(GooglePubsubAcknowledge::nack);
+
+ assertEquals(1, modifyAckDeadlineRequests.size());
+ ModifyAckDeadlineRequest nackRequest =
modifyAckDeadlineRequests.get(0);
+ assertEquals(List.of("ack-1", "ack-2", "ack-3"),
nackRequest.getAckIdsList());
+ assertEquals(0, nackRequest.getAckDeadlineSeconds());
+ assertEquals(SUBSCRIPTION, nackRequest.getSubscription());
+ assertTrue(ackRequests.isEmpty());
+ }
+
+ @Test
+ void ackIdOfAggregatedExchangeThatIsAlsoGroupedIsSentOnce() {
+ // an aggregation strategy may keep the first grouped exchange as the
aggregated one
+ CamelContext context = new DefaultCamelContext();
+ Exchange aggregated = exchange(context, "ack-1");
+ aggregated.getIn().setBody(List.of(aggregated, exchange(context,
"ack-2"), exchange(context, "ack-3")));
+
+ new AcknowledgeSync(this::stub, SUBSCRIPTION).ack(aggregated);
+
+ assertEquals(1, ackRequests.size());
+ assertEquals(List.of("ack-1", "ack-2", "ack-3"),
ackRequests.get(0).getAckIdsList());
+ }
+
+ @Test
+ void ackOfSingleExchangeAcksItsOwnMessage() {
+ CamelContext context = new DefaultCamelContext();
+ Exchange exchange = exchange(context, "ack-1");
+ exchange.getIn().setBody("hello".getBytes());
+
+ new AcknowledgeSync(this::stub, SUBSCRIPTION).ack(exchange);
+
+ assertEquals(1, ackRequests.size());
+ assertEquals(List.of("ack-1"), ackRequests.get(0).getAckIdsList());
+ }
+
+ @Test
+ void ackWithoutAnyAckIdSendsNoRequest() {
+ // e.g. the route removed the ack id header of the grouped exchanges
with removeHeaders("Camel*")
+ CamelContext context = new DefaultCamelContext();
+ Exchange aggregated = new DefaultExchange(context);
+ aggregated.getIn().setBody(List.of(new DefaultExchange(context), new
DefaultExchange(context)));
+
+ List<LogEvent> warnings = warningsOf(() -> new
AcknowledgeSync(this::stub, SUBSCRIPTION).ack(aggregated));
+
+ assertEquals(List.of(), ackRequests);
+ assertEquals(0, stubs.get());
+ assertEquals(1, warnings.size());
+
assertTrue(warnings.get(0).getMessage().getFormattedMessage().contains(aggregated.getExchangeId()));
+ }
+
+ @Test
+ void nackWithoutAckIdSendsNoRequest() {
+ CamelContext context = new DefaultCamelContext();
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("hello".getBytes());
+
+ List<LogEvent> warnings = warningsOf(() -> new
AcknowledgeSync(this::stub, SUBSCRIPTION).nack(exchange));
+
+ assertEquals(List.of(), modifyAckDeadlineRequests);
+ assertEquals(0, stubs.get());
+ assertEquals(1, warnings.size());
+
assertTrue(warnings.get(0).getMessage().getFormattedMessage().contains(exchange.getExchangeId()));
+ }
+
+ /**
+ * One pull returns three messages, the route aggregates them into one
exchange and acknowledges that exchange with
+ * the acknowledge header of the first grouped exchange.
+ */
+ private void routeAggregatedExchange(BiConsumer<GooglePubsubAcknowledge,
Exchange> acknowledgement) throws Exception {
+ SubscriberStub stub = stub();
+ 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&ackMode=NONE"
+ + "&maxMessagesPerPoll=10&maxDeliveryAttempts=0")
+ .aggregate(constant(true), new
GroupedExchangeAggregationStrategy()).completionSize(3)
+ .process(exchange -> {
+ List<Exchange> grouped =
exchange.getIn().getBody(List.class);
+ GooglePubsubAcknowledge acknowledge =
grouped.get(0).getIn().getHeader(
+
GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE, GooglePubsubAcknowledge.class);
+ acknowledgement.accept(acknowledge, exchange);
+ })
+ .to("mock:acknowledged");
+ }
+ });
+ MockEndpoint acknowledged =
context.getEndpoint("mock:acknowledged", MockEndpoint.class);
+ acknowledged.expectedMessageCount(1);
+
+ context.start();
+
+ acknowledged.assertIsSatisfied();
+ } finally {
+ context.stop();
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private SubscriberStub stub() {
+ stubs.incrementAndGet();
+ UnaryCallable<PullRequest, PullResponse> pullCallable =
mock(UnaryCallable.class);
+ when(pullCallable.futureCall(any(PullRequest.class)))
+
.thenReturn(ApiFutures.immediateFuture(PullResponse.newBuilder()
+ .addReceivedMessages(message("ack-1", "one"))
+ .addReceivedMessages(message("ack-2", "two"))
+ .addReceivedMessages(message("ack-3", "three"))
+ .build()))
+ // no more messages; cancelled when the consumer stops
+ .thenReturn(pending);
+
+ UnaryCallable<AcknowledgeRequest, Empty> acknowledgeCallable =
mock(UnaryCallable.class);
+
when(acknowledgeCallable.call(any(AcknowledgeRequest.class))).thenAnswer(invocation
-> {
+ ackRequests.add(invocation.getArgument(0));
+ return Empty.getDefaultInstance();
+ });
+
+ UnaryCallable<ModifyAckDeadlineRequest, Empty>
modifyAckDeadlineCallable = mock(UnaryCallable.class);
+
when(modifyAckDeadlineCallable.call(any(ModifyAckDeadlineRequest.class))).thenAnswer(invocation
-> {
+ modifyAckDeadlineRequests.add(invocation.getArgument(0));
+ return Empty.getDefaultInstance();
+ });
+
+ SubscriberStub stub = mock(SubscriberStub.class);
+ when(stub.pullCallable()).thenReturn(pullCallable);
+ when(stub.acknowledgeCallable()).thenReturn(acknowledgeCallable);
+
when(stub.modifyAckDeadlineCallable()).thenReturn(modifyAckDeadlineCallable);
+ return stub;
+ }
+
+ private static ReceivedMessage message(String ackId, String body) {
+ return ReceivedMessage.newBuilder()
+ .setAckId(ackId)
+ .setMessage(PubsubMessage.newBuilder()
+ .setMessageId(ackId)
+ .setData(ByteString.copyFromUtf8(body)))
+ .build();
+ }
+
+ private static Exchange exchange(CamelContext context, String ackId) {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(GooglePubsubConstants.ACK_ID, ackId);
+ return exchange;
+ }
+
+ /** Runs the action and returns the WARN events that {@link
AcknowledgeSync} logged meanwhile. */
+ private static List<LogEvent> warningsOf(Runnable action) {
+ Logger logger = (Logger) LogManager.getLogger(AcknowledgeSync.class);
+ LogEventCaptor captor = new LogEventCaptor();
+ logger.addAppender(captor);
+ try {
+ action.run();
+ } finally {
+ logger.removeAppender(captor);
+ }
+ return captor.events.stream().filter(event -> event.getLevel() ==
Level.WARN).toList();
+ }
+
+ private static final class LogEventCaptor extends AbstractAppender {
+ private final List<LogEvent> events = new CopyOnWriteArrayList<>();
+
+ LogEventCaptor() {
+ super("AcknowledgeSyncLogEventCaptor", null, null, false,
Property.EMPTY_ARRAY);
+ start();
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ events.add(event.toImmutable());
+ }
+ }
+}
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/PubsubTestSupport.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/PubsubTestSupport.java
index 500e6a920836..4a3ed37bf9ab 100644
---
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/PubsubTestSupport.java
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/PubsubTestSupport.java
@@ -25,6 +25,7 @@ import java.util.logging.LogManager;
import com.google.api.gax.core.CredentialsProvider;
import com.google.api.gax.core.NoCredentialsProvider;
import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
+import com.google.api.gax.rpc.AlreadyExistsException;
import com.google.api.gax.rpc.TransportChannelProvider;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
@@ -147,17 +148,21 @@ public class PubsubTestSupport extends CamelTestSupport {
public void createTopic(Topic topic) {
TopicAdminClient topicAdminClient = createTopicAdminClient();
-
- topicAdminClient.createTopic(topic);
-
+ try {
+ topicAdminClient.createTopic(topic);
+ } catch (AlreadyExistsException e) {
+ // created for an earlier test of the class, which shares the
emulator: safe to ignore
+ }
topicAdminClient.shutdown();
}
public void createSubscription(Subscription subscription) {
SubscriptionAdminClient subscriptionAdminClient =
createSubscriptionAdminClient();
-
- subscriptionAdminClient.createSubscription(subscription);
-
+ try {
+ subscriptionAdminClient.createSubscription(subscription);
+ } catch (AlreadyExistsException e) {
+ // created for an earlier test of the class, which shares the
emulator: safe to ignore
+ }
subscriptionAdminClient.shutdown();
}
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/integration/AggregatedManualAcknowledgementIT.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/integration/AggregatedManualAcknowledgementIT.java
new file mode 100644
index 000000000000..31e5c4b91c3d
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/integration/AggregatedManualAcknowledgementIT.java
@@ -0,0 +1,109 @@
+/*
+ * 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.integration;
+
+import java.util.List;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.google.pubsub.GooglePubsubConstants;
+import org.apache.camel.component.google.pubsub.PubsubTestSupport;
+import
org.apache.camel.component.google.pubsub.consumer.GooglePubsubAcknowledge;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy;
+import org.junit.jupiter.api.Test;
+
+/**
+ * With synchronous pull and ackMode=NONE, acknowledging an aggregated
exchange acknowledges every grouped message.
+ */
+public class AggregatedManualAcknowledgementIT extends PubsubTestSupport {
+
+ private static final String ACK_TOPIC = "aggregatedAckTopic";
+ private static final String ACK_SUBSCRIPTION = "aggregatedAckSubscription";
+ private static final String NACK_TOPIC = "aggregatedNackTopic";
+ private static final String NACK_SUBSCRIPTION =
"aggregatedNackSubscription";
+ private static final String CONSUMER_OPTIONS
+ =
"?synchronousPull=true&ackMode=NONE&maxMessagesPerPoll=10&maxDeliveryAttempts=0";
+
+ @EndpointInject("mock:ackReceived")
+ private MockEndpoint ackReceived;
+
+ @EndpointInject("mock:nackReceived")
+ private MockEndpoint nackReceived;
+
+ @Override
+ public void createTopicSubscription() {
+ createTopicSubscriptionPair(ACK_TOPIC, ACK_SUBSCRIPTION, 10);
+ // a long ack deadline: a redelivery during the test can only come
from the nack
+ createTopicSubscriptionPair(NACK_TOPIC, NACK_SUBSCRIPTION, 600);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("google-pubsub:{{project.id}}:" + ACK_SUBSCRIPTION +
CONSUMER_OPTIONS)
+ .routeId("ack").autoStartup(false)
+ .to("mock:ackReceived")
+ .aggregate(constant(true), new
GroupedExchangeAggregationStrategy()).completionSize(3)
+ .process(exchange ->
acknowledge(exchange).ack(exchange));
+
+ from("google-pubsub:{{project.id}}:" + NACK_SUBSCRIPTION +
CONSUMER_OPTIONS)
+ .routeId("nack").autoStartup(false)
+ .to("mock:nackReceived")
+ .aggregate(constant(true), new
GroupedExchangeAggregationStrategy()).completionSize(3)
+ .process(exchange ->
acknowledge(exchange).nack(exchange));
+ }
+ };
+ }
+
+ private static GooglePubsubAcknowledge acknowledge(Exchange aggregated) {
+ List<Exchange> grouped = aggregated.getIn().getBody(List.class);
+ return
grouped.get(0).getIn().getHeader(GooglePubsubConstants.GOOGLE_PUBSUB_ACKNOWLEDGE,
+ GooglePubsubAcknowledge.class);
+ }
+
+ @Test
+ public void ackOfAggregatedExchangeAcksEveryGroupedMessage() throws
Exception {
+ for (int i = 0; i < 3; i++) {
+ template.sendBody("google-pubsub:{{project.id}}:" + ACK_TOPIC,
"ack-" + i);
+ }
+ ackReceived.expectedBodiesReceivedInAnyOrder("ack-0", "ack-1",
"ack-2");
+ // the ack deadline of the subscription is 10 seconds: a message that
is not acked is delivered again by then
+ ackReceived.setAssertPeriod(15000);
+
+ context.getRouteController().startRoute("ack");
+
+ ackReceived.assertIsSatisfied();
+ }
+
+ @Test
+ public void nackOfAggregatedExchangeNacksEveryGroupedMessage() throws
Exception {
+ for (int i = 0; i < 3; i++) {
+ template.sendBody("google-pubsub:{{project.id}}:" + NACK_TOPIC,
"nack-" + i);
+ }
+ // the 3 messages are delivered again right after the nack, long
before their ack deadline
+ nackReceived.expectedMinimumMessageCount(6);
+
+ context.getRouteController().startRoute("nack");
+
+ nackReceived.assertIsSatisfied();
+ context.getRouteController().stopRoute("nack");
+ }
+}
