oscerd opened a new pull request, #1852: URL: https://github.com/apache/camel-kafka-connector/pull/1852
Fixes #1543. ### Problem A source connector whose route marshals its output publishes the Java object reference of a Camel stream instead of the payload: ``` org.apache.camel.converter.stream.InputStreamCache@6e37b6bd ``` With `ByteArrayConverter` the task fails instead (`ByteArrayConverter is not compatible with objects of type class org.apache.camel.converter.stream.InputStreamCache`). Camel Main enables stream caching by default, and with stream caching the output of a `marshal` step is a `StreamCache`. About 40 source Kamelets in the 4.22.0 catalog marshal their output, for example postgresql-source, mysql-source and salesforce-source. `CamelSourceTask` put that object into the `SourceRecord` as is. This still reproduces on 4.22.0 and `main`. ### Change - `CamelSourceTask` now reads a `StreamCache` (after resetting it), or any other `InputStream` body, into a `byte[]` with the Camel type converter before it builds the record. The value gets a `BYTES` schema, which is what the `marshal` step produces when stream caching is disabled. Other bodies are unchanged. - If the stream cannot be read, the task fails with a `ConnectException` instead of publishing an object reference. - `basic-configuration.adoc` gets a "Source Record Values" section. It says to use `ByteArrayConverter` to write these values unchanged, or `CamelTypeConverterTransform$Value` with `target.type=java.lang.String` to get a string. The generated example configurations use `StringConverter`, which writes a `byte[]` as `[B@...`. I checked this with the Kafka 4.2.1 converters. ### Behaviour change Records that used to carry a stream object now carry a `byte[]`. The standard converters could not handle the old values, so this only affects setups that worked around the bug with their own SMT expecting a `StreamCache`. ### Tests - `CamelSourceTaskTest#testSourcePollingWithMarshalledBody` uses `camel.source.marshal=syslog`, so the body goes through the same `marshal` step a Kamelet uses. It fails on `main` with `InputStreamCache@...`. - `CamelSourceTaskTest#testSourcePollingWithStreamBody` covers an `InputStreamCache` body and a `ByteArrayInputStream` body. It also fails on `main`. - `./mvnw -pl core -Psourcecheck install` runs 129 tests with 0 failures. The full reactor build `./mvnw clean verify -DskipTests` succeeds. _Claude Code on behalf of Andrea Cosentino_ 🤖 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]
