This is an automated email from the ASF dual-hosted git repository.
oscerd 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 4437dc45bce2 [backport camel-4.22.x] CAMEL-24350: camel-google-pubsub
- stop dropping list elements and null header values (#25618)
4437dc45bce2 is described below
commit 4437dc45bce219c396a0437d34d97858c3140770
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Aug 24 17:55:24 2026 +0200
[backport camel-4.22.x] CAMEL-24350: camel-google-pubsub - stop dropping
list elements and null header values (#25618)
CAMEL-24350: camel-google-pubsub - stop dropping list elements and null
header values (#25564)
A body holding both aggregated exchanges and plain payloads published only
the
exchanges and dropped the rest without a word. Such a list is neither of
the two
documented shapes, so it is now refused with a message saying what it
holds; a
list of only exchanges and a list used as a payload keep working as before.
Header pass-through also handed the message builder the value of every
header,
including the ones that do not convert to a string. A pubsub attribute
cannot be
null, so a single such header failed the send with a NullPointerException.
Those
headers are now skipped and logged at debug.
(cherry picked from commit 8f3edd566b2edc7690e569b3e0b6266c5edd61af)
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../camel-google/camel-google-pubsub/pom.xml | 5 ++
.../google/pubsub/GooglePubsubProducer.java | 27 +++++++---
.../unit/PubsubProducerBodyAndAttributesTest.java | 60 ++++++++++++++++++++++
3 files changed, 85 insertions(+), 7 deletions(-)
diff --git a/components/camel-google/camel-google-pubsub/pom.xml
b/components/camel-google/camel-google-pubsub/pom.xml
index 9033d44c912b..4fd784e73959 100644
--- a/components/camel-google/camel-google-pubsub/pom.xml
+++ b/components/camel-google/camel-google-pubsub/pom.xml
@@ -88,5 +88,10 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubProducer.java
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubProducer.java
index a07cbfc01593..d295dc870ab7 100644
---
a/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubProducer.java
+++
b/components/camel-google/camel-google-pubsub/src/main/java/org/apache/camel/component/google/pubsub/GooglePubsubProducer.java
@@ -70,15 +70,21 @@ public class GooglePubsubProducer extends DefaultProducer {
}
if (exchange.getIn().getBody() instanceof List) {
- boolean groupedExchanges = false;
- for (Object body : exchange.getIn().getBody(List.class)) {
- if (body instanceof Exchange) {
+ List<?> bodies = exchange.getIn().getBody(List.class);
+ long exchanges =
bodies.stream().filter(Exchange.class::isInstance).count();
+ if (exchanges == 0) {
+ // not an aggregated list: the list itself is the payload of a
single message
+ send(exchange);
+ } else if (exchanges == bodies.size()) {
+ for (Object body : bodies) {
send((Exchange) body);
- groupedExchanges = true;
}
- }
- if (!groupedExchanges) {
- send(exchange);
+ } else {
+ // a list holding both exchanges and plain payloads used to
publish only the exchanges and
+ // drop the rest without a word
+ throw new IllegalArgumentException(
+ "The body is a list mixing " + exchanges + "
exchange(s) with " + (bodies.size() - exchanges)
+ + " other element(s). Send
either an aggregated list of exchanges or a single payload.");
}
} else {
send(exchange);
@@ -129,6 +135,13 @@ public class GooglePubsubProducer extends DefaultProducer {
&&
headerFilterStrategy.applyFilterToCamelHeaders(camelHeader, value, exchange)) {
continue;
}
+ if (value == null) {
+ // a pubsub attribute cannot be null, and a header that does
not convert to a string is not
+ // one the message can carry
+ logger.debug("Skipping header {} as it has no string value to
send as a message attribute",
+ camelHeader);
+ continue;
+ }
messageBuilder.putAttributes(camelHeader, value);
}
diff --git
a/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/unit/PubsubProducerBodyAndAttributesTest.java
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/unit/PubsubProducerBodyAndAttributesTest.java
new file mode 100644
index 000000000000..14f26510c574
--- /dev/null
+++
b/components/camel-google/camel-google-pubsub/src/test/java/org/apache/camel/component/google/pubsub/unit/PubsubProducerBodyAndAttributesTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.unit;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.component.google.pubsub.GooglePubsubEndpoint;
+import org.apache.camel.component.google.pubsub.GooglePubsubProducer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Verifies what the producer does with a list body that is neither a list of
aggregated exchanges nor a payload.
+ */
+class PubsubProducerBodyAndAttributesTest {
+
+ private final GooglePubsubEndpoint endpoint = mock();
+ private final DefaultCamelContext context = new DefaultCamelContext();
+
+ @AfterEach
+ void tearDown() {
+ context.stop();
+ }
+
+ @Test
+ void aListMixingExchangesAndPayloadsIsRejected() {
+ GooglePubsubProducer producer = new GooglePubsubProducer(endpoint);
+
+ Exchange grouped = new DefaultExchange(context);
+ grouped.getIn().setBody("an aggregated exchange");
+
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody(List.of(grouped, "a plain payload"));
+
+ // publishing only the exchange and dropping the payload without a
word is worse than refusing the body
+ assertThatThrownBy(() -> producer.process(exchange))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("mixing 1 exchange(s) with 1 other
element(s)");
+ }
+}
