allthingssecurity opened a new pull request, #27798: URL: https://github.com/apache/camel/pull/27798
# Description [CAMEL-25611](https://issues.apache.org/jira/browse/CAMEL-25611) The insert, delete, update and upsert operations of `KuduProducer` ended with `connection.newSession().apply(operation);` and dropped the result. This PR fixes two defects of that line. **A row error was reported as a success.** A session of kudu-client has the flush mode `AUTO_FLUSH_SYNC` by default. In that mode `KuduSession.apply` sends the operation and returns its `OperationResponse`, and the client does not throw in two cases; it returns a response with a row error (`hasRowError()`): - a row that the tablet server rejected, for example an insert of a key that is already present, or an update or a delete of a row that does not exist; - a request that failed (read from kudu-client 1.18.1, not run against a Kudu server): `AsyncKuduSession$SingleOperationErrCallback`, the errback of the write in this flush mode, turns every `KuduException` of the request (a timeout, a tablet server that cannot be reached, a row outside the range partitions) into an `OperationResponse` whose `RowError` has the status of the exception. After such a failure it is not known whether the row was written. The producer did not look at the response, so in both cases the exchange completed without an exception and the route continued as if the row had been written. **The session of each exchange was never closed.** `AsyncKuduClient.newSession()` keeps every session in a set until the session is closed, so the client held one more session for every exchange, until the client itself was closed. The four operations now call one private method. It applies the operation, throws the new `KuduRowErrorException` ("Kudu reported a row error for the operation: ...") when the response has a row error, and closes the session. When the operation failed, a failure of the close is added to its exception as suppressed, so the first failure is the one that is reported. New public class: `KuduRowErrorException` (a `RuntimeCamelException`), with `getRowError()` for the `RowError` of the Kudu client, so that a route can tell a duplicate key from a missing row from a timeout by its status. The row error is not serialized with the exception. What changes for users (4.24 upgrade guide entry, and a new "Row errors" section in the component documentation): an insert of an existing key and an update or delete of a missing row now fail the exchange, and so does a write whose request failed. 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)`. A source that redelivers an insert now fails on the second delivery ("Already present"); `upsert` is the way out. How strict this is compared with other producers (read on main): - `MongoDbProducer` reports an update or a remove that matched nothing as a count and does not fail (`RECORDS_AFFECTED` / `RECORDS_MATCHED` at lines 518-519, `RECORDS_AFFECTED` at 533-535); `JdbcProducer` likewise sets the update count header (lines 206-208 and 276-278). A duplicate insert fails in MongoDB because the driver throws (`insertOne`, line 467). - So "insert of an existing key fails" matches these producers, and "update or delete of a missing row fails" is stricter than they are. I kept it because the Kudu client itself reports a missing row as an error, and the producer gives the route no other way to see it (there is no count header). - If failing an update or delete of a missing row is not wanted, the check could ignore the `NotFound` status of the row error for delete (and update). Such a limit has to be by status, not by operation: skipping the check for delete as a whole would hide its timeouts again. Another possible follow-up is an option for what Kudu offers for exactly this: the `*Ignore` operations (`newInsertIgnore`, `newUpdateIgnore`, `newDeleteIgnore`) or `setIgnoreAllDuplicateRows` / `setIgnoreAllNotFoundRows` on the session. Neither is part of this PR. Not changed: the table is still opened for every exchange (`openTable`), and the scan and create-table operations are as on main. Tests: - New `KuduProducerRowErrorTest` (10 tests) with a mocked Kudu client; the module gets `mockito-core` as a test dependency for it. The Kudu harness that the other tests of the module use only runs on Linux x86_64, and this test also has to make the session answer with a row error. For each of the four operations: a response with a row error fails the exchange with `KuduRowErrorException` (same `RowError`, the step after the producer gets nothing) and the session is closed; a response without a row error completes the exchange and the session is closed. A response whose row error has the status "Timed out" (the shape the client gives a failed request) fails the exchange with a `KuduRowErrorException` that exposes that status. One more test: when `apply` throws, the session is closed, the exception of `apply` is the exception of the exchange, and a failure of the close is its suppressed exception. - Without the change (the producer of main, same test) all ten fail: `Unexpected null value, expected: <org.apache.camel.component.kudu.KuduRowErrorException> but was: <null>` for the five row-error cases (four operations and the timed-out request), `Wanted but not invoked: kuduSession.close()` for the four applied-row cases, and `array lengths differ, expected: <1> but was: <0>` for the test where `apply` throws (the failure of the close is not attached, because the session is not closed). - With the change, on macOS arm64 with `-Dmaven.test.skip=false` (the module skips its tests on this platform by default): 24 run, 0 failures, 12 skipped. The 12 skipped tests are the existing ones that need the Kudu harness; I could not run them here. I read them: none of them provokes a row error. They skip by assumption when the harness cannot start (`Assumptions.assumeTrue(ikc.hasKuduHarness(), ...)` in `AbstractKuduTest`), and the open ticket [CAMEL-23493](https://issues.apache.org/jira/browse/CAMEL-23493) reports that this is what happens to most of them, so the mocked test may be the only one that exercises these producer operations. What was not run: nothing was run against a Kudu server. That the server answers a duplicate key and a missing row with a row error, and does not throw, is taken from the code of kudu-client 1.18.1 (`OperationResponse`, `RowError`, the default flush mode of `AsyncKuduSession`) and from its documentation. That a failed request (timeout, unreachable tablet server) comes back as a row error is taken from the code of kudu-client 1.18.1 only; the test for it uses a mocked response of that shape. How it was found: a sweep of the producers for results of the remote API that are never inspected, and a small TLA+ model (a table, exchanges with one row operation each, the sessions of the client). The model is only a sketch of the code. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). (The ticket is drafted and not filed yet; CAMEL-25611 is a placeholder until it is.) # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built and tested camel-kudu, including the formatter and import-sort plugins, and copied the changed component page to the catalog. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
