This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new a34118e864 Fix #651: add a data formats page to the user guide (#1858)
a34118e864 is described below
commit a34118e86413982aa7aed95a8ae0d695deb3adc7
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 11:37:06 2026 +0100
Fix #651: add a data formats page to the user guide (#1858)
The camel.source/sink.marshal and unmarshal options were only listed in
the basic configuration table, where the source rows had been given the
sink wording.
- Add user-guide/data-formats.adoc: what the four options do, where they
run in the route CamelKafkaConnectMain builds, the Camel 4 data format
names, how to make a data format available to a connector, how to
configure it (camel.dataformat.<name>.<option> or a bean), what the
source-side record value is, and an example.
- Link it from the nav and from the basic configuration table, and fix
the description of the source options there.
- Fix CAMEL_SOURCE_MARSHAL_DOC, which said "unmarshal".
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../kafkaconnector/CamelSourceConnectorConfig.java | 2 +-
docs/modules/ROOT/nav.adoc | 1 +
.../ROOT/pages/user-guide/basic-configuration.adoc | 8 +-
.../ROOT/pages/user-guide/data-formats.adoc | 203 +++++++++++++++++++++
4 files changed, 209 insertions(+), 5 deletions(-)
diff --git
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
index 8132ace6fb..357dc43928 100644
---
a/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
+++
b/core/src/main/java/org/apache/camel/kafkaconnector/CamelSourceConnectorConfig.java
@@ -30,7 +30,7 @@ public class CamelSourceConnectorConfig extends
CamelConnectorConfig {
public static final String CAMEL_SOURCE_MARSHAL_DEFAULT = null;
public static final String CAMEL_SOURCE_MARSHAL_CONF =
"camel.source.marshal";
- public static final String CAMEL_SOURCE_MARSHAL_DOC = "The camel
dataformat name to use to unmarshal data to the topic";
+ public static final String CAMEL_SOURCE_MARSHAL_DOC = "The camel
dataformat name to use to marshal data to the topic";
public static final String CAMEL_SOURCE_URL_DEFAULT = null;
public static final String CAMEL_SOURCE_URL_CONF = "camel.source.url";
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index 659f8d37ba..eac60d93cf 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -7,6 +7,7 @@
***
xref:user-guide/getting-started/getting-started-with-packages.adoc[Packages
documentation]
** xref:user-guide/basic-concepts.adoc[Basic concepts]
** xref:user-guide/basic-configuration.adoc[Basic configuration]
+** xref:user-guide/data-formats.adoc[Data formats]
** xref:user-guide/aggregation.adoc[Aggregation]
** xref:user-guide/idempotency.adoc[Idempotency]
** xref:user-guide/remove-headers.adoc[Remove Headers]
diff --git a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
index b645387bba..c257210167 100644
--- a/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
+++ b/docs/modules/ROOT/pages/user-guide/basic-configuration.adoc
@@ -8,8 +8,8 @@ For a Sink connector the basic options are:
[width="100%",cols="2,5,^2,1",options="header"]
|===
| Name | Description | Default | Priority
-| camel.sink.marshal | The camel dataformat name to use to marshal data to the
destination | null | HIGH
-| camel.sink.unmarshal | The camel dataformat name to use to unmarshal data
from the topic | null | HIGH
+| camel.sink.marshal | The camel dataformat name to use to marshal data to the
destination, see xref:user-guide/data-formats.adoc[Data formats] | null | HIGH
+| camel.sink.unmarshal | The camel dataformat name to use to unmarshal data
from the topic, see xref:user-guide/data-formats.adoc[Data formats] | null |
HIGH
| camel.sink.contentLogLevel | Log level for the record's content. Valid
values: TRACE, DEBUG, INFO, WARN, ERROR, OFF. | OFF | HIGH
| camel.beans.aggregate | A reference to an aggregate bean, in the form of
#class: | null | MEDIUM
| camel.aggregation.size | The size of the aggregation, to be used in
combination with camel.beans.aggregate | 10 | MEDIUM
@@ -34,8 +34,8 @@ For a Source connector the basic options are:
[width="100%",cols="2,5,^2,1",options="header"]
|===
| Name | Description | Default | Priority
-| camel.source.marshal | The camel dataformat name to use to marshal data to
the destination | null | HIGH
-| camel.source.unmarshal | The camel dataformat name to use to unmarshal data
from the topic | null | HIGH
+| camel.source.marshal | The camel dataformat name to use to marshal data to
the topic, see xref:user-guide/data-formats.adoc[Data formats] | null | HIGH
+| camel.source.unmarshal | The camel dataformat name to use to unmarshal data
from the source, see xref:user-guide/data-formats.adoc[Data formats] | null |
HIGH
| camel.source.contentLogLevel | Log level for the record's content. Valid
values: TRACE, DEBUG, INFO, WARN, ERROR, OFF. | OFF | HIGH
| camel.source.maxBatchPollSize | The max number of messages retrieved in a
single poll() | 1000L | MEDIUM
| camel.source.maxPollDuration | The maximum time in milliseconds spent in a
single call to poll() | 1000L | MEDIUM
diff --git a/docs/modules/ROOT/pages/user-guide/data-formats.adoc
b/docs/modules/ROOT/pages/user-guide/data-formats.adoc
new file mode 100644
index 0000000000..ee6514e2e9
--- /dev/null
+++ b/docs/modules/ROOT/pages/user-guide/data-formats.adoc
@@ -0,0 +1,203 @@
+[[DataFormats-DataFormats]]
+= Data Formats
+
+A Camel https://camel.apache.org/manual/data-format.html[data format] converts
a message body between a serialized
+form, such as JSON, CSV, HL7 or syslog text, and another representation,
usually Java objects. To _marshal_ is to
+write the body in the serialized form, to _unmarshal_ is to parse the
serialized form.
+
+Every connector, Kamelet-based or component-based, can apply a data format to
the body of each message it moves,
+with configuration only.
+
+NOTE: A data format is not a Kafka Connect converter. The `key.converter` and
`value.converter` of a connector
+translate between the bytes stored in Kafka and the Kafka Connect record. A
data format works on the Camel message
+body inside the connector: on a sink connector it runs after the converter and
the SMTs have produced the record
+value, on a source connector it runs before the SMTs and the converter.
+
+[[DataFormats-Options]]
+== Options
+
+[width="100%",cols="2,5",options="header"]
+|===
+| Name | Description
+| camel.sink.marshal | The data format used to marshal the record value before
it is sent to the destination.
+| camel.sink.unmarshal | The data format used to unmarshal the record value
before it is sent to the destination.
+| camel.source.marshal | The data format used to marshal the data received
from the source before it is written to the topic.
+| camel.source.unmarshal | The data format used to unmarshal the data received
from the source before it is written to the topic.
+|===
+
+The value is the name of a data format, for example `jackson` (see
<<DataFormats-Names>>), or the name of a data
+format bean (see <<DataFormats-Configuring>>). None of these options is set by
default, except in
+`camel-syslog-kafka-connector`, which sets `camel.source.unmarshal=syslog` and
`camel.sink.marshal=syslog`.
+
+[[DataFormats-Route]]
+== Where the data formats run
+
+Each connector task builds a single Camel route from its configuration. The
data formats run at a fixed place in that
+route, together with the other optional stages; a stage that is not configured
is left out.
+
+On a sink connector:
+
+. the record value becomes the message body, and by default the record headers
starting with `CamelHeader.` or
+ `CamelProperty.` become Camel headers or exchange properties (see
+
xref:user-guide/basic-configuration.adoc#BasicConfiguration-MessageLevelHeaders[Message
Level Headers]);
+. `camel.remove.headers.pattern` removes the matching headers (see
xref:user-guide/remove-headers.adoc[Remove Headers]);
+. `camel.sink.marshal`;
+. `camel.sink.unmarshal`;
+. the aggregation, when `camel.beans.aggregate` is set (see
xref:user-guide/aggregation.adoc[Aggregation]);
+. the idempotent consumer, when `camel.idempotency.enabled=true` (see
xref:user-guide/idempotency.adoc[Idempotency]);
+. the message is sent to the destination.
+
+On a source connector:
+
+. the message is received from the source;
+. `camel.source.marshal`;
+. `camel.source.unmarshal`;
+. the aggregation, when `camel.beans.aggregate` is set;
+. the idempotent consumer, when `camel.idempotency.enabled=true`;
+. `camel.remove.headers.pattern` removes the matching headers;
+. the message body becomes the record value, and by default the headers become
record headers.
+
+As a consequence:
+
+* when both options of a connector are set, marshal always runs before
unmarshal, in both directions;
+* the aggregation strategy and the idempotency expression
(`camel.idempotency.expression.type=body`) see the body
+ produced by the data formats, not the original one;
+* on a sink connector the headers removed by `camel.remove.headers.pattern`
never reach the data formats, while on a
+ source connector the data formats still see every header.
+
+If a data format cannot process a body, for example
`camel.sink.unmarshal=jackson` on a record value that is not
+JSON, the message fails like any other delivery error.
+
+[[DataFormats-Names]]
+== Data format names
+
+The value is the name Camel 4 registers the data format under, not the name of
its Maven artifact. Some common ones:
+
+[width="100%",cols="2,2,2",options="header"]
+|===
+| Data format | Name | Artifact
+| JSON (Jackson) | `jackson` | `camel-jackson`
+| XML (Jackson) | `jacksonXml` | `camel-jacksonxml`
+| Avro (Jackson) | `avroJackson` | `camel-jackson-avro`
+| Protobuf (Jackson) | `protobufJackson` | `camel-jackson-protobuf`
+| JSON (Gson) | `gson` | `camel-gson`
+| Avro | `avro` | `camel-avro`
+| CSV | `csv` | `camel-csv`
+| Base64 | `base64` | `camel-base64`
+| HL7 | `hl7` | `camel-hl7`
+| Syslog | `syslog` | `camel-syslog`
+| YAML (SnakeYAML) | `snakeYaml` | `camel-snakeyaml`
+| Zip file | `zipFile` | `camel-zipfile`
+| GZip deflater | `gzipDeflater` | `camel-zip-deflater`
+| JAXB | `jaxb` | `camel-jaxb`
+|===
+
+The Jackson 3 modules (`camel-jackson3`, `camel-jackson3xml`,
`camel-jackson3-avro` and `camel-jackson3-protobuf`)
+register the same names as their Jackson 2 counterparts.
+
+Older configurations can carry names that no longer exist: the JSON Jackson
data format, for example, is `jackson`,
+while Camel 3.14 and earlier registered it as `json-jackson`. An unknown name
makes the connector task fail to start
+with `Cannot find data format in registry with ref: <name>`.
+
+The https://camel.apache.org/components/next/dataformats/index.html[Camel data
formats] page lists every data format
+with its options.
+
+[[DataFormats-Classpath]]
+== Making the data format available
+
+The core `camel-kafka-connector` jar and its dependencies include no data
format. The data format you name must be on
+the classpath of the connector: its jar, and the jars it depends on, built for
the Camel version of the connector,
+have to be in the connector's directory under the Kafka Connect `plugin.path`.
+
+Some connectors include a data format already. The component-based connectors,
such as `camel-file-kafka-connector`,
+`camel-https-kafka-connector` or `camel-jdbc-kafka-connector`, and many
Kamelet-based connectors, such as
+`camel-cassandra-sink-kafka-connector`, ship `camel-jackson`;
`camel-syslog-kafka-connector` ships `camel-syslog`.
+Look for the jar of the data format, for example
`camel-jackson-<version>.jar`, in the connector's directory.
+
+To add a data format to a connector, copy its jars into the connector's
directory, or build an extended connector
+package with the
xref:user-guide/extending-connector/archetype-dataformat-connector.adoc[Extensible
Dataformat
+connector archetype], which adds the data format as a dependency of the
connector.
+
+[[DataFormats-Configuring]]
+== Configuring the data format
+
+Set the data format options with `camel.dataformat.<name>.<option>`
properties, where `<name>` is the data format name
+and `<option>` one of the options listed on its page in the Camel
documentation:
+
+[source,properties]
+----
+camel.sink.marshal=jackson
+camel.dataformat.jackson.prettyPrint=true
+----
+
+These properties configure the single instance Camel creates for that name,
which every marshal or unmarshal option
+naming it uses.
+A `camel.dataformat.<name>.*` property makes the task fail to start when the
data format `<name>` is not on the
+classpath, even when no marshal or unmarshal option uses it.
+
+To configure an instance under a name of your choice, for example to marshal
and unmarshal with different settings,
+declare the data format as a bean and use the bean name as the option value:
+
+[source,properties]
+----
+camel.beans.prettyJson=#class:org.apache.camel.component.jackson.JacksonDataFormat
+camel.beans.prettyJson.prettyPrint=true
+camel.sink.marshal=prettyJson
+----
+
+`#class:` creates a new instance of the class, and the
`camel.beans.<name>.<property>` entries set its properties;
+see the Camel https://camel.apache.org/manual/property-binding.html[property
binding] documentation for the syntax.
+
+Some data formats can let a message header choose the Java type to unmarshal
to, for example Jackson with
+`allowUnmarshallType=true` (the `CamelJacksonUnmarshalType` header) or
`allowJmsType=true` (the `JMSType` header).
+Both options are disabled by default. On a sink connector a record header
`CamelHeader.<name>` becomes the Camel
+header `<name>`, so keep these options disabled when the topic is written by
producers you do not trust, or remove
+those headers with `camel.remove.headers.pattern` (see the
xref:security-model.adoc[Security Model]).
+
+[[DataFormats-SourceMarshal]]
+== Marshalling on a source connector
+
+The output of `camel.source.marshal` becomes the record value as a `byte[]`.
Use the `ByteArrayConverter` to write
+it to the topic unchanged:
+
+[source,properties]
+----
+camel.source.marshal=jackson
+value.converter=org.apache.kafka.connect.converters.ByteArrayConverter
+----
+
+The Kafka `StringConverter` writes a `byte[]` value as a Java object reference
(`[B@...`). To get a string value,
+for example for JSON, convert the value before the converter runs with the
`CamelTypeConverterTransform` SMT that
+every connector includes:
+
+[source,properties]
+----
+camel.source.marshal=jackson
+transforms=toString
+transforms.toString.type=org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value
+transforms.toString.target.type=java.lang.String
+value.converter=org.apache.kafka.connect.storage.StringConverter
+----
+
+[[DataFormats-Example]]
+== Example
+
+This sink connector writes each record of the `orders` topic, which holds JSON
objects, to its own file as indented
+JSON. The `JsonConverter` without schemas turns each record value into a
`java.util.Map`, and `camel.sink.marshal`
+writes it back as JSON. `camel-file-kafka-connector` already ships
`camel-jackson`.
+
+[source,properties]
+----
+name=orders-to-files
+connector.class=org.apache.camel.kafkaconnector.file.CamelFileSinkConnector
+tasks.max=1
+topics=orders
+
+key.converter=org.apache.kafka.connect.storage.StringConverter
+value.converter=org.apache.kafka.connect.json.JsonConverter
+value.converter.schemas.enable=false
+
+camel.sink.path.directoryName=/tmp/orders
+camel.sink.marshal=jackson
+camel.dataformat.jackson.prettyPrint=true
+----
