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 f25f798b0828 CAMEL-25611: camel-kudu - Fail the exchange when Kudu
answers a row operation with a row error (#27798)
f25f798b0828 is described below
commit f25f798b0828be6b5fe55bc37637092fbeb4b706
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 14 17:16:23 2026 +0530
CAMEL-25611: camel-kudu - Fail the exchange when Kudu answers a row
operation with a row error (#27798)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/kudu-component.adoc | 11 ++
components/camel-kudu/pom.xml | 6 +
.../camel-kudu/src/main/docs/kudu-component.adoc | 11 ++
.../apache/camel/component/kudu/KuduProducer.java | 41 ++++-
.../component/kudu/KuduRowErrorException.java | 53 ++++++
.../component/kudu/KuduProducerRowErrorTest.java | 200 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 15 ++
7 files changed, 333 insertions(+), 4 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kudu-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kudu-component.adoc
index 9f9ddea80cc8..c2173f4327f9 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kudu-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kudu-component.adoc
@@ -55,3 +55,14 @@ This map will represent a row of the table whose elements
are columns, where the
The output body format will be a `java.util.List<java.util.Map<String,
Object>>`.
Each element of the list will be a different row of the table.
Each row is a `Map<String, Object>` whose elements will be each pair of column
name and column value for that row.
+
+=== Row errors
+
+The producer fails the exchange with a `KuduRowErrorException` when the answer
to an insert, delete, update or upsert
+carries a row error. Its `getRowError()` returns the row error of the Kudu
client, whose status tells the kind of
+error. There are two cases:
+
+* a row that Kudu rejected: an insert of a row whose key is already in the
table, or an update or a delete of a row
+that is not in the table. The row was not written. Use the `upsert` operation
to write a row whether it exists or not.
+* a request that failed, for example because it timed out or because no tablet
server could be reached. The Kudu
+client reports it the same way. In this case it is not known whether the row
was written.
diff --git a/components/camel-kudu/pom.xml b/components/camel-kudu/pom.xml
index ac127ed973b5..dabf8fc5a85a 100644
--- a/components/camel-kudu/pom.xml
+++ b/components/camel-kudu/pom.xml
@@ -104,6 +104,12 @@
<artifactId>junit-jupiter-migrationsupport</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
<!-- Spin a kudu cluster to run integration tests -->
<!-- as described in
diff --git a/components/camel-kudu/src/main/docs/kudu-component.adoc
b/components/camel-kudu/src/main/docs/kudu-component.adoc
index 9f9ddea80cc8..c2173f4327f9 100644
--- a/components/camel-kudu/src/main/docs/kudu-component.adoc
+++ b/components/camel-kudu/src/main/docs/kudu-component.adoc
@@ -55,3 +55,14 @@ This map will represent a row of the table whose elements
are columns, where the
The output body format will be a `java.util.List<java.util.Map<String,
Object>>`.
Each element of the list will be a different row of the table.
Each row is a `Map<String, Object>` whose elements will be each pair of column
name and column value for that row.
+
+=== Row errors
+
+The producer fails the exchange with a `KuduRowErrorException` when the answer
to an insert, delete, update or upsert
+carries a row error. Its `getRowError()` returns the row error of the Kudu
client, whose status tells the kind of
+error. There are two cases:
+
+* a row that Kudu rejected: an insert of a row whose key is already in the
table, or an update or a delete of a row
+that is not in the table. The row was not written. Use the `upsert` operation
to write a row whether it exists or not.
+* a request that failed, for example because it timed out or because no tablet
server could be reached. The Kudu
+client reports it the same way. In this case it is not known whether the row
was written.
diff --git
a/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduProducer.java
b/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduProducer.java
index 7c20f2602e7d..6ef2ea99d7ad 100644
---
a/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduProducer.java
+++
b/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduProducer.java
@@ -30,7 +30,10 @@ import org.apache.kudu.client.Insert;
import org.apache.kudu.client.KuduClient;
import org.apache.kudu.client.KuduException;
import org.apache.kudu.client.KuduPredicate;
+import org.apache.kudu.client.KuduSession;
import org.apache.kudu.client.KuduTable;
+import org.apache.kudu.client.Operation;
+import org.apache.kudu.client.OperationResponse;
import org.apache.kudu.client.PartialRow;
import org.apache.kudu.client.Update;
import org.apache.kudu.client.Upsert;
@@ -103,7 +106,7 @@ public class KuduProducer extends DefaultProducer {
}
}
- connection.newSession().apply(insert);
+ apply(connection, insert);
}
private void doDelete(Exchange exchange, String tableName) throws
KuduException {
@@ -124,7 +127,7 @@ public class KuduProducer extends DefaultProducer {
}
}
- connection.newSession().apply(delete);
+ apply(connection, delete);
}
private void doUpdate(Exchange exchange, String tableName) throws
KuduException {
@@ -144,7 +147,7 @@ public class KuduProducer extends DefaultProducer {
}
}
- connection.newSession().apply(update);
+ apply(connection, update);
}
private void doUpsert(Exchange exchange, String tableName) throws
KuduException {
@@ -164,7 +167,37 @@ public class KuduProducer extends DefaultProducer {
}
}
- connection.newSession().apply(upsert);
+ apply(connection, upsert);
+ }
+
+ /**
+ * Applies one row operation with a session of its own and closes the
session.
+ * <p>
+ * A session has the flush mode AUTO_FLUSH_SYNC by default: the operation
is sent before the call returns, and the
+ * Kudu client does not throw but returns a response with a row error in
two cases: for a row that the tablet server
+ * rejected (the key is already present, the row is not found, ...), and
for a request that failed (it timed out, no
+ * tablet server could be reached, ...). In the second case it is not
known whether the row was written. Both fail
+ * the exchange with a {@link KuduRowErrorException}.
+ * <p>
+ * A session stays registered with its client until it is closed. After an
operation without a row error a failure
+ * of the close is thrown; in this flush mode the close has nothing left
to send.
+ */
+ private static void apply(KuduClient connection, Operation operation)
throws KuduException {
+ KuduSession session = connection.newSession();
+ try {
+ OperationResponse response = session.apply(operation);
+ if (response != null && response.hasRowError()) {
+ throw new KuduRowErrorException(response.getRowError());
+ }
+ } catch (KuduException | RuntimeException e) {
+ try {
+ session.close();
+ } catch (KuduException | RuntimeException closeFailure) {
+ e.addSuppressed(closeFailure);
+ }
+ throw e;
+ }
+ session.close();
}
private void doCreateTable(Exchange exchange, String tableName) throws
KuduException {
diff --git
a/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduRowErrorException.java
b/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduRowErrorException.java
new file mode 100644
index 000000000000..d1cb251d62e1
--- /dev/null
+++
b/components/camel-kudu/src/main/java/org/apache/camel/component/kudu/KuduRowErrorException.java
@@ -0,0 +1,53 @@
+/*
+ * 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.kudu;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.kudu.client.RowError;
+
+/**
+ * Thrown by the producer when the response of an insert, delete, update or
upsert carries a row error. There are two
+ * cases, and the {@link RowError#getErrorStatus() status} of the row error
tells them apart:
+ * <ul>
+ * <li>a row that the tablet server rejected, for example because its key is
already present (insert) or because the row
+ * is not found (update, delete): the row was not written;</li>
+ * <li>a request that failed, for example because it timed out or because no
tablet server could be reached. The Kudu
+ * client reports such a request the same way in the default flush mode of a
session. In this case it is not known
+ * whether the row was written.</li>
+ * </ul>
+ * The message holds the status only. It does not include the row, because the
text of a {@link RowError} prints every
+ * column value of the row; {@link #getRowError()} gives access to it.
+ */
+public class KuduRowErrorException extends RuntimeCamelException {
+
+ private static final long serialVersionUID = 1L;
+
+ private final transient RowError rowError;
+
+ public KuduRowErrorException(RowError rowError) {
+ super("Kudu reported a row error for the operation: " +
rowError.getErrorStatus());
+ this.rowError = rowError;
+ }
+
+ /**
+ * The row error of the Kudu client; its {@link RowError#getErrorStatus()
status} tells the kind of error (already
+ * present, not found, timed out, network error, ...). It is not
serialized with the exception.
+ */
+ public RowError getRowError() {
+ return rowError;
+ }
+}
diff --git
a/components/camel-kudu/src/test/java/org/apache/camel/component/kudu/KuduProducerRowErrorTest.java
b/components/camel-kudu/src/test/java/org/apache/camel/component/kudu/KuduProducerRowErrorTest.java
new file mode 100644
index 000000000000..031175554fbe
--- /dev/null
+++
b/components/camel-kudu/src/test/java/org/apache/camel/component/kudu/KuduProducerRowErrorTest.java
@@ -0,0 +1,200 @@
+/*
+ * 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.kudu;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.kudu.ColumnSchema;
+import org.apache.kudu.Schema;
+import org.apache.kudu.Type;
+import org.apache.kudu.client.Delete;
+import org.apache.kudu.client.Insert;
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduSession;
+import org.apache.kudu.client.KuduTable;
+import org.apache.kudu.client.Operation;
+import org.apache.kudu.client.OperationResponse;
+import org.apache.kudu.client.RowError;
+import org.apache.kudu.client.Status;
+import org.apache.kudu.client.Update;
+import org.apache.kudu.client.Upsert;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+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.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Runs the producer against a mocked Kudu client, so it does not need the
Kudu harness of the other tests of this
+ * module. The session answers as a session with the default flush mode does:
a row that the tablet server rejected, and
+ * a request that failed (a timeout for example, after which it is not known
whether the row was written), are the row
+ * error of the returned response, not an exception.
+ */
+class KuduProducerRowErrorTest extends CamelTestSupport {
+
+ private final KuduClient client = mock(KuduClient.class);
+ private final KuduSession session = mock(KuduSession.class);
+ private final KuduTable table = mock(KuduTable.class);
+ private final OperationResponse response = mock(OperationResponse.class);
+ private final RowError rowError = mock(RowError.class);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ Schema schema = new Schema(
+ Arrays.asList(
+ new ColumnSchema.ColumnSchemaBuilder("id",
Type.INT32).key(true).build(),
+ new ColumnSchema.ColumnSchemaBuilder("name",
Type.STRING).build()));
+
+ when(client.openTable("TestTable")).thenReturn(table);
+ when(client.newSession()).thenReturn(session);
+ when(table.getSchema()).thenReturn(schema);
+ when(table.newInsert()).thenAnswer(invocation ->
operation(Insert.class, schema));
+ when(table.newDelete()).thenAnswer(invocation ->
operation(Delete.class, schema));
+ when(table.newUpdate()).thenAnswer(invocation ->
operation(Update.class, schema));
+ when(table.newUpsert()).thenAnswer(invocation ->
operation(Upsert.class, schema));
+ when(session.apply(any(Operation.class))).thenReturn(response);
+ when(response.getRowError()).thenReturn(rowError);
+ when(rowError.getErrorStatus()).thenReturn(Status.AlreadyPresent("key
already present"));
+ when(rowError.toString())
+ .thenReturn("Row error for row=(int32 id=5, string
name=secret), status=Already present: key already present");
+
+ CamelContext context = super.createCamelContext();
+ KuduComponent kudu = new KuduComponent();
+ kudu.setKuduClient(client);
+ context.addComponent("kudu", kudu);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ for (String operation : new String[] { "insert", "delete",
"update", "upsert" }) {
+ from("direct:" + operation)
+ .to("kudu:localhost:7051/TestTable?operation=" +
operation)
+ .to("mock:result");
+ }
+ }
+ };
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "insert", "delete", "update", "upsert" })
+ void rowErrorFailsTheExchange(String operation) throws Exception {
+ when(response.hasRowError()).thenReturn(true);
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:" + operation, e ->
e.getIn().setBody(row()));
+
+ KuduRowErrorException failure =
assertInstanceOf(KuduRowErrorException.class, exchange.getException(),
+ "exception of an exchange whose " + operation + " was answered
with a row error");
+ assertSame(rowError, failure.getRowError());
+ // the status only: the text of the row error prints the column values
of the row
+ assertEquals("Kudu reported a row error for the operation: Already
present: key already present",
+ failure.getMessage());
+ result.assertIsSatisfied();
+ verify(session, times(1)).close();
+ }
+
+ @Test
+ void failedRequestFailsTheExchangeWithItsStatus() throws Exception {
+ // what the Kudu client returns for a request that timed out: a
response whose row error has the status of
+ // the failure
+ Status timedOut = Status.TimedOut("can not complete before timeout");
+ when(response.hasRowError()).thenReturn(true);
+ when(rowError.getErrorStatus()).thenReturn(timedOut);
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:delete", e ->
e.getIn().setBody(row()));
+
+ KuduRowErrorException failure =
assertInstanceOf(KuduRowErrorException.class, exchange.getException(),
+ "exception of an exchange whose delete was answered with the
row error of a request that timed out");
+ assertSame(timedOut, failure.getRowError().getErrorStatus());
+ assertTrue(failure.getRowError().getErrorStatus().isTimedOut());
+ assertEquals("Kudu reported a row error for the operation: " +
timedOut, failure.getMessage());
+ result.assertIsSatisfied();
+ verify(session, times(1)).close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "insert", "delete", "update", "upsert" })
+ void sessionIsClosedAfterAnAppliedRow(String operation) throws Exception {
+ when(response.hasRowError()).thenReturn(false);
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(1);
+
+ Exchange exchange = template.send("direct:" + operation, e ->
e.getIn().setBody(row()));
+
+ assertNull(exchange.getException());
+ result.assertIsSatisfied();
+ verify(session, times(1)).apply(any(Operation.class));
+ verify(session, times(1)).close();
+ }
+
+ @Test
+ void sessionIsClosedWhenTheOperationFails() throws Exception {
+ IllegalStateException failure = new IllegalStateException("apply
failed");
+ IllegalStateException closeFailure = new IllegalStateException("close
failed too");
+ when(session.apply(any(Operation.class))).thenThrow(failure);
+ when(session.close()).thenThrow(closeFailure);
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedMessageCount(0);
+
+ Exchange exchange = template.send("direct:insert", e ->
e.getIn().setBody(row()));
+
+ // the failure of the operation is reported, not the failure of the
close that follows it
+ assertSame(failure, exchange.getException());
+ assertArrayEquals(new Throwable[] { closeFailure },
failure.getSuppressed());
+ result.assertIsSatisfied();
+ verify(session, times(1)).close();
+ verify(response, never()).hasRowError();
+ }
+
+ private static Map<String, Object> row() {
+ Map<String, Object> row = new HashMap<>();
+ row.put("id", 5);
+ row.put("name", "Samuel");
+ return row;
+ }
+
+ private static <T extends Operation> T operation(Class<T> type, Schema
schema) {
+ T operation = mock(type);
+ when(operation.getRow()).thenReturn(schema.newPartialRow());
+ return operation;
+ }
+}
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 35c93c0ad8e8..81ba4af58a03 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
@@ -269,6 +269,21 @@ poll returns only the tweets newer than the last one
consumed, and a page has `c
default of 15 applied before). Raise `count` and `numberOfPages` if a search
can match more tweets than that between
two polls. A consumer with `type=direct` returned nothing when keywords were
given, and now runs the search at startup.
+=== camel-kudu - a row error of Kudu fails the exchange
+
+The `insert`, `delete`, `update` and `upsert` operations of the producer
ignored the answer of Kudu. The Kudu client
+does not throw for a row that Kudu rejected; it returns a row error, for
example for an insert of a key that is already
+present, or for an update or a delete of a row that is not found. Such an
exchange completed as if the row had been
+written. A write that timed out or could not reach a tablet server was
reported the same way, and completed the
+exchange too; for such a write it is not known whether the row was written.
+The producer now fails the exchange in both cases with a
`KuduRowErrorException` (`getRowError()` returns the row
+error of the Kudu client, whose status tells the kind of error). Routes that
relied on rejected rows being skipped
+silently can use the `upsert` operation, or handle the exception, for example
with
+`onException(KuduRowErrorException.class).handled(true)`.
+
+The producer also closes the Kudu session of each operation. The sessions
stayed registered with the Kudu client until
+the client was closed, one more for every exchange.
+
=== camel-atom - sortEntries sorts by the updated date and routes the oldest
entry first
With `sortEntries=true` the `atom` consumer sorted the entries of a feed by
their published date and routed them from
