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 ac961f81d7cb CAMEL-26117: camel-debezium - the row flattened by
ExtractNewRecordState is the body, and mongodb updates expose their
updateDescription (#28893)
ac961f81d7cb is described below
commit ac961f81d7cba1cb040cecabffd311169f20c054
Author: Claus Ibsen <[email protected]>
AuthorDate: Mon Nov 2 20:36:12 2026 +0100
CAMEL-26117: camel-debezium - the row flattened by ExtractNewRecordState is
the body, and mongodb updates expose their updateDescription (#28893)
…ate is the body, and mongodb updates expose their updateDescription
With the ExtractNewRecordState (or ExtractNewDocumentState) SMT the record
value is the flattened
row, which has no after or op field, so the body and the
CamelDebeziumOperation header were null.
A Struct value without an after field (that is not a Debezium schema
change, heartbeat or
transaction event) is now the body, and the operation is read from the __op
field or header the
SMT adds with add.fields=op or add.headers=op.
debezium-mongodb fell back to the patch field, which Debezium 2 removed; an
update without the
full document lost its changes. The updateDescription is now in the
CamelDebeziumUpdateDescription
header. The Struct to Map converter also converts Structs inside lists and
maps.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/components/debezium-mongodb.json | 5 +-
.../camel/catalog/docs/debezium-db2-component.adoc | 6 +
.../catalog/docs/debezium-mongodb-component.adoc | 10 ++
.../catalog/docs/debezium-mysql-component.adoc | 6 +
.../catalog/docs/debezium-oracle-component.adoc | 6 +
.../catalog/docs/debezium-postgres-component.adoc | 6 +
.../catalog/docs/debezium-sqlserver-component.adoc | 6 +
.../component/debezium/DebeziumConstants.java | 5 +
.../camel/component/debezium/DebeziumEndpoint.java | 38 ++++++-
.../component/debezium/DebeziumTypeConverter.java | 34 ++++--
.../component/debezium/DebeziumEndpointTest.java | 95 +++++++++++++++-
.../debezium/DebeziumTypeConverterTest.java | 27 +++++
.../src/main/docs/debezium-db2-component.adoc | 6 +
.../debezium/mongodb/debezium-mongodb.json | 5 +-
.../src/main/docs/debezium-mongodb-component.adoc | 10 ++
.../debezium/mongodb/DebeziumMongodbEndpoint.java | 28 +++--
.../mongodb/DebeziumMongodbEndpointTest.java | 125 +++++++++++++++++++++
.../src/main/docs/debezium-mysql-component.adoc | 6 +
.../src/main/docs/debezium-oracle-component.adoc | 6 +
.../src/main/docs/debezium-postgres-component.adoc | 6 +
.../main/docs/debezium-sqlserver-component.adoc | 6 +
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 14 +++
.../dsl/DebeziumMongodbEndpointBuilderFactory.java | 15 +++
23 files changed, 439 insertions(+), 32 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/debezium-mongodb.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/debezium-mongodb.json
index 82dbf9653fd4..f0c2bbd3b8a6 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/debezium-mongodb.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/debezium-mongodb.json
@@ -122,10 +122,11 @@
"CamelDebeziumOperation": { "index": 3, "kind": "header", "displayName":
"", "group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the type of event operation. Values for the
connector are c for create (or insert), u for update, d for delete or r for
read (in the case of a initial sync) or in case of a snapshot event.",
"constantName": "org.apache.camel.c [...]
"CamelDebeziumTimestamp": { "index": 4, "kind": "header", "displayName":
"", "group": "consumer", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the time (using the system clock in the
JVM) at which the connector processed the event.", "constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_TIMESTAMP" },
"CamelDebeziumBefore": { "index": 5, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "Struct",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, contains the state of the row before the
event occurred.", "constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_BEFORE" },
- "CamelDebeziumDdlSQL": { "index": 6, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the ddl sql text of the event.",
"constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_DDL_SQL" }
+ "CamelDebeziumDdlSQL": { "index": 6, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the ddl sql text of the event.",
"constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_DDL_SQL" },
+ "CamelDebeziumUpdateDescription": { "index": 7, "kind": "header",
"displayName": "", "group": "consumer", "label": "", "required": false,
"javaType": "Map<String, Object>", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "If presents, the
description of an update: updatedFields (the changed fields as a JSON String),
removedFields (the names of the removed fields) and truncatedArrays (the arrays
that were truncated, with their field and [...]
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "null" ], "javaType":
"java.lang.String", "format": "json", "dependsOn": [ "additionalProperties" ],
"description": "The document after the change as a JSON String (MongoDB
extended JSON): unmarshal it to work with its fields. No body for a delete or a
tombstone, nor for an update when captureMode does not look up the full
document or the document is gone by then; the operation and the key are in
headers" }
+ "consumerOut": { "carriers": [ "text", "null" ], "javaType":
"java.lang.String", "format": "json", "dependsOn": [ "additionalProperties" ],
"description": "The document after the change as a JSON String (MongoDB
extended JSON): unmarshal it to work with its fields. No body for a delete or a
tombstone, nor for an update when captureMode does not look up the full
document or the document is gone by then; the operation and the key are in
headers, and the changed fields of an update in t [...]
},
"properties": {
"name": { "index": 0, "kind": "path", "displayName": "Name", "group":
"consumer", "label": "consumer", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.debezium.mongodb.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration",
"configurationField": "configuration", "description": "Unique name for the
connector. Attempting to registe [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-db2-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-db2-component.adoc
index d2a57da69f31..694709d2ff5a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-db2-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-db2-component.adoc
@@ -80,6 +80,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mongodb-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mongodb-component.adoc
index 85c243958bd3..95bbd7b3a316 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mongodb-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mongodb-component.adoc
@@ -83,6 +83,16 @@
https://debezium.io/documentation/reference/0.10/connectors/mongodb.html#connect
If the message body is not `null` (in case of tombstones), it contains the
state of the row after the event occurred. It is in `String` JSON format, and
you can unmarshal using Camel JSON Data Format.
+An update has no body when `captureMode` does not look up the full document
(such as `change_streams`). The changed
+fields of an update are in the `CamelDebeziumUpdateDescription` header, as a
`Map` with `updatedFields` (the changed
+fields as a JSON String), `removedFields` (the names of the removed fields)
and `truncatedArrays`.
+
+With the Debezium `ExtractNewDocumentState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState`),
the
+event is flattened to the document, and the body is that document as a
`Struct`. The `CamelDebeziumOperation` header is
+then set only when the transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mysql-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mysql-component.adoc
index 34b0b811980c..ff87d1557167 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mysql-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-mysql-component.adoc
@@ -91,6 +91,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-oracle-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-oracle-component.adoc
index ea3bb07fbefe..d2555e6dcf69 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-oracle-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-oracle-component.adoc
@@ -78,6 +78,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-postgres-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-postgres-component.adoc
index 7efd8adcf172..0f7e6e9de9a6 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-postgres-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-postgres-component.adoc
@@ -78,6 +78,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-sqlserver-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-sqlserver-component.adoc
index 96bc0187996e..43913d21386c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-sqlserver-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/debezium-sqlserver-component.adoc
@@ -76,6 +76,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConstants.java
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConstants.java
index 6a7d289aa269..93d0fceecf14 100644
---
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConstants.java
+++
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumConstants.java
@@ -47,6 +47,11 @@ public final class DebeziumConstants {
public static final String HEADER_BEFORE = HEADER_PREFIX + "Before";
@Metadata(description = "If presents, the ddl sql text of the event.",
javaType = "String")
public static final String HEADER_DDL_SQL = HEADER_PREFIX + "DdlSQL";
+ @Metadata(description = "If presents, the description of an update:
`updatedFields` (the changed fields as a JSON String), "
+ + "`removedFields` (the names of the removed
fields) and `truncatedArrays` (the arrays that were "
+ + "truncated, with their `field` and new `size`).",
+ javaType = "Map<String, Object>", applicableFor =
"debezium-mongodb")
+ public static final String HEADER_UPDATE_DESCRIPTION = HEADER_PREFIX +
"UpdateDescription";
private DebeziumConstants() {
}
diff --git
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumEndpoint.java
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumEndpoint.java
index 463104457556..39373159058c 100644
---
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumEndpoint.java
+++
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumEndpoint.java
@@ -31,6 +31,7 @@ import org.apache.camel.support.DefaultEndpoint;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.header.Header;
import org.apache.kafka.connect.source.SourceRecord;
/**
@@ -38,6 +39,10 @@ import org.apache.kafka.connect.source.SourceRecord;
*/
public abstract class DebeziumEndpoint<C extends
EmbeddedDebeziumConfiguration> extends DefaultEndpoint {
+ private static final String DEBEZIUM_SCHEMA_NAME_PREFIX = "io.debezium.";
+ // the name of the field or header that the ExtractNewRecordState SMT adds
for the operation with its default prefix
+ private static final String FLATTENED_OPERATION = "__op";
+
protected DebeziumEndpoint(String uri, DebeziumComponent<C> component) {
super(uri, component);
}
@@ -79,7 +84,7 @@ public abstract class DebeziumEndpoint<C extends
EmbeddedDebeziumConfiguration>
// extract values from SourceRecord
final Map<String, Object> sourceMetadata =
extractSourceMetadataValueFromValueStruct(valueSchema, value);
- final Object operation = extractValueFromValueStruct(valueSchema,
value, Envelope.FieldName.OPERATION);
+ final Object operation = extractOperation(sourceRecord, valueSchema,
value);
final Object before = extractValueFromValueStruct(valueSchema, value,
Envelope.FieldName.BEFORE);
final Object body = extractBodyValueFromValueStruct(valueSchema,
value);
final Object timestamp = extractValueFromValueStruct(valueSchema,
value, Envelope.FieldName.TIMESTAMP);
@@ -112,12 +117,39 @@ public abstract class DebeziumEndpoint<C extends
EmbeddedDebeziumConfiguration>
// first we try with normal extraction from value struct
final Object valueExtracted = extractValueFromValueStruct(schema,
value, fieldName);
- if (valueExtracted == null && !isSchemaAStructSchema(schema)) { // we
could have anything other than struct, we just return that
+ // we could have anything other than struct, or the flattened row of
the ExtractNewRecordState SMT, we just return that
+ if (valueExtracted == null && (!isSchemaAStructSchema(schema) ||
isFlattenedRecordValue(schema, value))) {
return value;
}
return valueExtracted;
}
+ /**
+ * Whether the value is a row that a single message transformation such as
ExtractNewRecordState (or
+ * ExtractNewDocumentState for MongoDB) flattened out of the change event
envelope: a Struct without an
+ * {@code after} field, that is not one of the Debezium events such as a
schema change, a heartbeat or a transaction
+ * metadata event (their schema names start with {@code io.debezium.}).
+ */
+ private boolean isFlattenedRecordValue(final Schema schema, final Object
value) {
+ return value != null && isSchemaAStructSchema(schema)
+ && schema.field(Envelope.FieldName.AFTER) == null
+ && (schema.name() == null ||
!schema.name().startsWith(DEBEZIUM_SCHEMA_NAME_PREFIX));
+ }
+
+ private Object extractOperation(final SourceRecord sourceRecord, final
Schema schema, final Object value) {
+ final Object operation = extractValueFromValueStruct(schema, value,
Envelope.FieldName.OPERATION);
+ if (operation != null || !isFlattenedRecordValue(schema, value)) {
+ return operation;
+ }
+ // the ExtractNewRecordState SMT adds the operation as the __op field
(add.fields=op) or header (add.headers=op)
+ final Object flattenedOperation = extractValueFromValueStruct(schema,
value, FLATTENED_OPERATION);
+ if (flattenedOperation != null) {
+ return flattenedOperation;
+ }
+ final Header header =
sourceRecord.headers().lastWithName(FLATTENED_OPERATION);
+ return header != null ? header.value() : null;
+ }
+
private Map<String, Object>
extractSourceMetadataValueFromValueStruct(final Schema schema, final Object
value) {
// we want to convert metadata to map since it facilitate usage and
also struct structure is not needed for the metadata
final Object valueExtracted = extractValueFromValueStruct(schema,
value, Envelope.FieldName.SOURCE);
@@ -128,7 +160,7 @@ public abstract class DebeziumEndpoint<C extends
EmbeddedDebeziumConfiguration>
return null;
}
- private Object extractValueFromValueStruct(final Schema schema, final
Object value, final String fieldName) {
+ protected Object extractValueFromValueStruct(final Schema schema, final
Object value, final String fieldName) {
// first we check if we have a value and a schema of struct type
if (isSchemaAStructSchema(schema) && value != null) {
// now we return our desired fieldName
diff --git
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumTypeConverter.java
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumTypeConverter.java
index 867d10d1130f..a9bac9778c38 100644
---
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumTypeConverter.java
+++
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/main/java/org/apache/camel/component/debezium/DebeziumTypeConverter.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.debezium;
+import java.util.ArrayList;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.camel.Converter;
@@ -29,8 +31,9 @@ public final class DebeziumTypeConverter {
}
/**
- * Convert {@link Struct} to {@link HashMap}, this only works with flat
fields and it doesn't handle nested
- * structure. Also as a result of the conversion, the schema data will be
lost which is expected.
+ * Convert {@link Struct} to {@link HashMap}. A nested {@link Struct} is
converted to a {@link Map} too, also inside
+ * a list (an array field) or as the value of a map (a map field). As a
result of the conversion, the schema data
+ * will be lost which is expected.
*
* @param struct
* @return {@link Map}
@@ -39,17 +42,24 @@ public final class DebeziumTypeConverter {
public static Map<String, Object> toMap(final Struct struct) {
final HashMap<String, Object> fieldsToValues = new HashMap<>();
- struct.schema().fields().forEach(field -> {
- Object value = struct.get(field);
-
- // recursive call if we have nested structs
- if (value instanceof Struct) {
- fieldsToValues.put(field.name(), toMap((Struct) value));
- } else {
- fieldsToValues.put(field.name(), value);
- }
- });
+ struct.schema().fields().forEach(field ->
fieldsToValues.put(field.name(), convertValue(struct.get(field))));
return fieldsToValues;
}
+
+ private static Object convertValue(final Object value) {
+ // recursive call if we have nested structs
+ if (value instanceof Struct struct) {
+ return toMap(struct);
+ } else if (value instanceof List<?> list) {
+ final List<Object> answer = new ArrayList<>(list.size());
+ list.forEach(element -> answer.add(convertValue(element)));
+ return answer;
+ } else if (value instanceof Map<?, ?> map) {
+ final Map<Object, Object> answer = new HashMap<>();
+ map.forEach((key, element) -> answer.put(convertValue(key),
convertValue(element)));
+ return answer;
+ }
+ return value;
+ }
}
diff --git
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
index 1a6994117f37..3223ffe60df5 100644
---
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
+++
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumEndpointTest.java
@@ -193,10 +193,77 @@ public class DebeziumEndpointTest {
assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
assertNull(inMessage.getHeader(DebeziumConstants.HEADER_KEY));
+ // a Struct that is not a change event envelope is the body, as the
row flattened by the ExtractNewRecordState SMT
+ final Struct body = (Struct) inMessage.getBody();
+ assertNotNull(body);
+ assertEquals((byte) 1, body.getInt8("id").byteValue());
+ }
+
+ @Test
+ void testIfCreatesExchangeFromFlattenedRecord() {
+ final SourceRecord sourceRecord = createFlattenedRecord(false, null);
+
+ final Exchange exchange = debeziumEndpoint.createDbzExchange(null,
sourceRecord);
+ final Message inMessage = exchange.getIn();
+
+ // assert headers
+ assertEquals("dummy",
inMessage.getHeader(DebeziumConstants.HEADER_IDENTIFIER));
+ final Struct key = (Struct)
inMessage.getHeader(DebeziumConstants.HEADER_KEY);
+ assertEquals(12345, key.getInt32("id").intValue());
+ assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+
// assert value
final Struct body = (Struct) inMessage.getBody();
- // we have to get value of after with struct, we are strict about this
case
- assertNull(body);
+ assertNotNull(body);
+ assertEquals((byte) 1, body.getInt8("id").byteValue());
+ assertEquals("jane", body.getString("name"));
+ }
+
+ @Test
+ void testIfCreatesExchangeFromFlattenedRecordWithOperationField() {
+ final SourceRecord sourceRecord = createFlattenedRecord(true, null);
+
+ final Exchange exchange = debeziumEndpoint.createDbzExchange(null,
sourceRecord);
+ final Message inMessage = exchange.getIn();
+
+ assertEquals(Envelope.Operation.CREATE.code(),
inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+ final Struct body = (Struct) inMessage.getBody();
+ assertNotNull(body);
+ assertEquals("jane", body.getString("name"));
+ }
+
+ @Test
+ void testIfCreatesExchangeFromFlattenedRecordWithOperationHeader() {
+ final SourceRecord sourceRecord = createFlattenedRecord(false,
Envelope.Operation.UPDATE.code());
+
+ final Exchange exchange = debeziumEndpoint.createDbzExchange(null,
sourceRecord);
+ final Message inMessage = exchange.getIn();
+
+ assertEquals(Envelope.Operation.UPDATE.code(),
inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+ final Struct body = (Struct) inMessage.getBody();
+ assertNotNull(body);
+ assertEquals("jane", body.getString("name"));
+ }
+
+ @Test
+ void testIfCreatesExchangeFromSchemaChangeRecord() {
+ final Schema schemaChangeSchema =
SchemaBuilder.struct().name("io.debezium.connector.mysql.SchemaChangeValue")
+ .field("databaseName", SchemaBuilder.string())
+ .field("ddl", SchemaBuilder.string())
+ .build();
+ final Struct schemaChange = new Struct(schemaChangeSchema);
+ schemaChange.put("databaseName", "inventory");
+ schemaChange.put("ddl", "CREATE TABLE users (id INT)");
+ final SourceRecord sourceRecord = new SourceRecord(
+ new HashMap<>(), createSourceOffset(), "dummy", null, null,
schemaChangeSchema, schemaChange);
+
+ final Exchange exchange = debeziumEndpoint.createDbzExchange(null,
sourceRecord);
+ final Message inMessage = exchange.getIn();
+
+ // a Debezium schema change event is not a row, the DDL is in its
header
+ assertEquals("CREATE TABLE users (id INT)",
inMessage.getHeader(DebeziumConstants.HEADER_DDL_SQL));
+ assertNull(inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+ assertNull(inMessage.getBody());
}
@Test
@@ -288,6 +355,30 @@ public class DebeziumEndpointTest {
return new SourceRecord(new HashMap<>(), new HashMap<>(), "dummy",
recordSchema, before);
}
+ private SourceRecord createFlattenedRecord(final boolean
withOperationField, final String operationHeader) {
+ // the row of a change event flattened by the ExtractNewRecordState
SMT, with add.fields=op or add.headers=op
+ final SchemaBuilder builder =
SchemaBuilder.struct().name("dummy.inventory.users.Value")
+ .field("id", SchemaBuilder.int8())
+ .field("name", SchemaBuilder.string());
+ if (withOperationField) {
+ builder.field("__op", SchemaBuilder.string());
+ }
+ final Schema recordSchema = builder.build();
+ final Struct row = new Struct(recordSchema);
+ row.put("id", (byte) 1);
+ row.put("name", "jane");
+ if (withOperationField) {
+ row.put("__op", Envelope.Operation.CREATE.code());
+ }
+ final SourceRecord sourceRecord = new SourceRecord(
+ new HashMap<>(), createSourceOffset(), "dummy",
createKeySchema(),
+ createKeyRecord(), recordSchema, row);
+ if (operationHeader != null) {
+ sourceRecord.headers().addString("__op", operationHeader);
+ }
+ return sourceRecord;
+ }
+
private SourceRecord createStringRecord() {
final Schema recordSchema = Schema.STRING_SCHEMA;
return new SourceRecord(new HashMap<>(), createSourceOffset(),
"dummy", recordSchema, "test_record");
diff --git
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumTypeConverterTest.java
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumTypeConverterTest.java
index 8381b4f697ea..fde7b1000263 100644
---
a/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumTypeConverterTest.java
+++
b/components/camel-debezium/camel-debezium-common/camel-debezium-common-component/src/test/java/org/apache/camel/component/debezium/DebeziumTypeConverterTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.debezium;
+import java.util.List;
import java.util.Map;
import org.apache.kafka.connect.data.Schema;
@@ -71,6 +72,32 @@ public class DebeziumTypeConverterTest {
assertEquals(30, ((Map) outputValue.get("details")).get("age"));
}
+ @Test
+ void testToMapStructsInListAndMap() {
+ Schema itemSchema = SchemaBuilder.struct().field("sku",
SchemaBuilder.STRING_SCHEMA).build();
+
+ Schema valueSchema = SchemaBuilder.struct()
+ .field("items", SchemaBuilder.array(itemSchema).build())
+ .field("byName",
SchemaBuilder.map(SchemaBuilder.STRING_SCHEMA, itemSchema).build())
+ .field("tags",
SchemaBuilder.array(SchemaBuilder.STRING_SCHEMA).build())
+ .build();
+
+ Struct inputValue = new Struct(valueSchema)
+ .put("items", List.of(new Struct(itemSchema).put("sku", "a1"),
new Struct(itemSchema).put("sku", "b2")))
+ .put("byName", Map.of("first", new
Struct(itemSchema).put("sku", "c3")))
+ .put("tags", List.of("x", "y"));
+
+ final Map<String, Object> outputValue =
DebeziumTypeConverter.toMap(inputValue);
+
+ final List<?> items = (List<?>) outputValue.get("items");
+ assertEquals(2, items.size());
+ assertEquals("a1", ((Map<?, ?>) items.get(0)).get("sku"));
+ assertEquals("b2", ((Map<?, ?>) items.get(1)).get("sku"));
+ final Map<?, ?> byName = (Map<?, ?>) outputValue.get("byName");
+ assertEquals("c3", ((Map<?, ?>) byName.get("first")).get("sku"));
+ assertEquals(List.of("x", "y"), outputValue.get("tags"));
+ }
+
private Struct createTestStruct(final int id, final String name, final
boolean valid) {
final Schema schema = SchemaBuilder.struct()
.field("id", Schema.INT32_SCHEMA)
diff --git
a/components/camel-debezium/camel-debezium-db2/src/main/docs/debezium-db2-component.adoc
b/components/camel-debezium/camel-debezium-db2/src/main/docs/debezium-db2-component.adoc
index d2a57da69f31..694709d2ff5a 100644
---
a/components/camel-debezium/camel-debezium-db2/src/main/docs/debezium-db2-component.adoc
+++
b/components/camel-debezium/camel-debezium-db2/src/main/docs/debezium-db2-component.adoc
@@ -80,6 +80,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-mongodb/src/generated/resources/META-INF/org/apache/camel/component/debezium/mongodb/debezium-mongodb.json
b/components/camel-debezium/camel-debezium-mongodb/src/generated/resources/META-INF/org/apache/camel/component/debezium/mongodb/debezium-mongodb.json
index 82dbf9653fd4..f0c2bbd3b8a6 100644
---
a/components/camel-debezium/camel-debezium-mongodb/src/generated/resources/META-INF/org/apache/camel/component/debezium/mongodb/debezium-mongodb.json
+++
b/components/camel-debezium/camel-debezium-mongodb/src/generated/resources/META-INF/org/apache/camel/component/debezium/mongodb/debezium-mongodb.json
@@ -122,10 +122,11 @@
"CamelDebeziumOperation": { "index": 3, "kind": "header", "displayName":
"", "group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the type of event operation. Values for the
connector are c for create (or insert), u for update, d for delete or r for
read (in the case of a initial sync) or in case of a snapshot event.",
"constantName": "org.apache.camel.c [...]
"CamelDebeziumTimestamp": { "index": 4, "kind": "header", "displayName":
"", "group": "consumer", "label": "", "required": false, "javaType": "Long",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the time (using the system clock in the
JVM) at which the connector processed the event.", "constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_TIMESTAMP" },
"CamelDebeziumBefore": { "index": 5, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "Struct",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, contains the state of the row before the
event occurred.", "constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_BEFORE" },
- "CamelDebeziumDdlSQL": { "index": 6, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the ddl sql text of the event.",
"constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_DDL_SQL" }
+ "CamelDebeziumDdlSQL": { "index": 6, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "deprecationNote": "", "autowired": false, "secret":
false, "description": "If presents, the ddl sql text of the event.",
"constantName":
"org.apache.camel.component.debezium.DebeziumConstants#HEADER_DDL_SQL" },
+ "CamelDebeziumUpdateDescription": { "index": 7, "kind": "header",
"displayName": "", "group": "consumer", "label": "", "required": false,
"javaType": "Map<String, Object>", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "description": "If presents, the
description of an update: updatedFields (the changed fields as a JSON String),
removedFields (the names of the removed fields) and truncatedArrays (the arrays
that were truncated, with their field and [...]
},
"bodyTypes": {
- "consumerOut": { "carriers": [ "text", "null" ], "javaType":
"java.lang.String", "format": "json", "dependsOn": [ "additionalProperties" ],
"description": "The document after the change as a JSON String (MongoDB
extended JSON): unmarshal it to work with its fields. No body for a delete or a
tombstone, nor for an update when captureMode does not look up the full
document or the document is gone by then; the operation and the key are in
headers" }
+ "consumerOut": { "carriers": [ "text", "null" ], "javaType":
"java.lang.String", "format": "json", "dependsOn": [ "additionalProperties" ],
"description": "The document after the change as a JSON String (MongoDB
extended JSON): unmarshal it to work with its fields. No body for a delete or a
tombstone, nor for an update when captureMode does not look up the full
document or the document is gone by then; the operation and the key are in
headers, and the changed fields of an update in t [...]
},
"properties": {
"name": { "index": 0, "kind": "path", "displayName": "Name", "group":
"consumer", "label": "consumer", "required": true, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "configurationClass":
"org.apache.camel.component.debezium.mongodb.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration",
"configurationField": "configuration", "description": "Unique name for the
connector. Attempting to registe [...]
diff --git
a/components/camel-debezium/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
b/components/camel-debezium/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
index 85c243958bd3..95bbd7b3a316 100644
---
a/components/camel-debezium/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
+++
b/components/camel-debezium/camel-debezium-mongodb/src/main/docs/debezium-mongodb-component.adoc
@@ -83,6 +83,16 @@
https://debezium.io/documentation/reference/0.10/connectors/mongodb.html#connect
If the message body is not `null` (in case of tombstones), it contains the
state of the row after the event occurred. It is in `String` JSON format, and
you can unmarshal using Camel JSON Data Format.
+An update has no body when `captureMode` does not look up the full document
(such as `change_streams`). The changed
+fields of an update are in the `CamelDebeziumUpdateDescription` header, as a
`Map` with `updatedFields` (the changed
+fields as a JSON String), `removedFields` (the names of the removed fields)
and `truncatedArrays`.
+
+With the Debezium `ExtractNewDocumentState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState`),
the
+event is flattened to the document, and the body is that document as a
`Struct`. The `CamelDebeziumOperation` header is
+then set only when the transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-mongodb/src/main/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpoint.java
b/components/camel-debezium/camel-debezium-mongodb/src/main/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpoint.java
index 33107405abff..b69d5dd1883c 100644
---
a/components/camel-debezium/camel-debezium-mongodb/src/main/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpoint.java
+++
b/components/camel-debezium/camel-debezium-mongodb/src/main/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpoint.java
@@ -18,10 +18,13 @@ package org.apache.camel.component.debezium.mongodb;
import java.util.Map;
-import io.debezium.data.Envelope;
+import io.debezium.connector.mongodb.MongoDbFieldName;
import org.apache.camel.Category;
+import org.apache.camel.Exchange;
import org.apache.camel.component.debezium.DebeziumConstants;
+import org.apache.camel.component.debezium.DebeziumConsumer;
import org.apache.camel.component.debezium.DebeziumEndpoint;
+import org.apache.camel.component.debezium.DebeziumTypeConverter;
import
org.apache.camel.component.debezium.mongodb.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration;
import org.apache.camel.spi.BodyType;
import org.apache.camel.spi.BodyType.Carrier;
@@ -29,7 +32,8 @@ import org.apache.camel.spi.BodyTypes;
import org.apache.camel.spi.EndpointServiceLocation;
import org.apache.camel.spi.UriEndpoint;
import org.apache.camel.spi.UriParam;
-import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
/**
* Capture changes from a MongoDB database.
@@ -43,7 +47,8 @@ import org.apache.kafka.connect.data.Schema;
+ " unmarshal it to work with
its fields. No body for a delete or a tombstone,"
+ " nor for an update when
captureMode does not look up the full document or the"
+ " document is gone by then;"
- + " the operation and the key
are in headers"))
+ + " the operation and the key
are in headers, and the changed fields of an"
+ + " update in the
CamelDebeziumUpdateDescription header"))
public final class DebeziumMongodbEndpoint extends
DebeziumEndpoint<MongoDbConnectorEmbeddedDebeziumConfiguration>
implements EndpointServiceLocation {
@@ -88,14 +93,15 @@ public final class DebeziumMongodbEndpoint extends
DebeziumEndpoint<MongoDbConne
}
@Override
- protected Object extractBodyValueFromValueStruct(Schema schema, Object
value) {
- // according to DBZ docs, `after` field only presents on the create
events, however for updates there is field `patch`
- //
https://debezium.io/documentation/reference/0.10/connectors/mongodb.html
- // hence first we test for `after`, if null we return `patch` instead
- final Object after = extractFieldValueFromValueStruct(schema, value,
Envelope.FieldName.AFTER);
- if (after != null) {
- return after;
+ public Exchange createDbzExchange(final DebeziumConsumer consumer, final
SourceRecord sourceRecord) {
+ final Exchange exchange = super.createDbzExchange(consumer,
sourceRecord);
+ // an update has the changed fields in updateDescription, and no
`after` value unless captureMode looks up the
+ // full document
+ final Object updateDescription =
extractValueFromValueStruct(sourceRecord.valueSchema(), sourceRecord.value(),
+ MongoDbFieldName.UPDATE_DESCRIPTION);
+ if (updateDescription instanceof Struct struct) {
+
exchange.getIn().setHeader(DebeziumConstants.HEADER_UPDATE_DESCRIPTION,
DebeziumTypeConverter.toMap(struct));
}
- return extractFieldValueFromValueStruct(schema, value, "patch");
+ return exchange;
}
}
diff --git
a/components/camel-debezium/camel-debezium-mongodb/src/test/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpointTest.java
b/components/camel-debezium/camel-debezium-mongodb/src/test/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpointTest.java
new file mode 100644
index 000000000000..919b1406701f
--- /dev/null
+++
b/components/camel-debezium/camel-debezium-mongodb/src/test/java/org/apache/camel/component/debezium/mongodb/DebeziumMongodbEndpointTest.java
@@ -0,0 +1,125 @@
+/*
+ * 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.debezium.mongodb;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.debezium.connector.mongodb.MongoDbFieldName;
+import io.debezium.connector.mongodb.MongoDbSchemaFactory;
+import io.debezium.data.Envelope;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.debezium.DebeziumConstants;
+import
org.apache.camel.component.debezium.mongodb.configuration.MongoDbConnectorEmbeddedDebeziumConfiguration;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class DebeziumMongodbEndpointTest {
+
+ private static final Schema ENVELOPE_SCHEMA =
SchemaBuilder.struct().name("dummy.inventory.customers.Envelope")
+ .field(Envelope.FieldName.AFTER, Schema.OPTIONAL_STRING_SCHEMA)
+ .field(MongoDbFieldName.UPDATE_DESCRIPTION,
MongoDbSchemaFactory.get().updatedDescriptionSchema())
+ .field(Envelope.FieldName.OPERATION, Schema.STRING_SCHEMA)
+ .field(Envelope.FieldName.TIMESTAMP, Schema.OPTIONAL_INT64_SCHEMA)
+ .build();
+
+ private DebeziumMongodbEndpoint endpoint;
+
+ @BeforeEach
+ public void setUp() {
+ endpoint = new DebeziumMongodbEndpoint(
+ "debezium-mongodb:dummy", new DebeziumMongodbComponent(new
DefaultCamelContext()),
+ new MongoDbConnectorEmbeddedDebeziumConfiguration());
+ }
+
+ @Test
+ void testIfCreatesExchangeFromCreateRecord() {
+ final Struct value = new Struct(ENVELOPE_SCHEMA)
+ .put(Envelope.FieldName.AFTER, "{\"_id\": 1, \"name\":
\"jane\"}")
+ .put(Envelope.FieldName.OPERATION,
Envelope.Operation.CREATE.code())
+ .put(Envelope.FieldName.TIMESTAMP, 1234L);
+
+ final Message inMessage = endpoint.createDbzExchange(null,
createRecord(ENVELOPE_SCHEMA, value)).getIn();
+
+ assertEquals(Envelope.Operation.CREATE.code(),
inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+ assertEquals("{\"_id\": 1, \"name\": \"jane\"}", inMessage.getBody());
+
assertNull(inMessage.getHeader(DebeziumConstants.HEADER_UPDATE_DESCRIPTION));
+ }
+
+ @Test
+ void testIfCreatesExchangeFromUpdateRecordWithoutFullDocument() {
+ final Schema updateDescriptionSchema =
ENVELOPE_SCHEMA.field(MongoDbFieldName.UPDATE_DESCRIPTION).schema();
+ final Schema truncatedArraySchema =
updateDescriptionSchema.field(MongoDbFieldName.TRUNCATED_ARRAYS).schema()
+ .valueSchema();
+ final Struct updateDescription = new Struct(updateDescriptionSchema)
+ .put(MongoDbFieldName.UPDATED_FIELDS, "{\"name\": \"john\"}")
+ .put(MongoDbFieldName.REMOVED_FIELDS, List.of("nickname"))
+ .put(MongoDbFieldName.TRUNCATED_ARRAYS, List.of(new
Struct(truncatedArraySchema)
+ .put(MongoDbFieldName.ARRAY_FIELD_NAME, "tags")
+ .put(MongoDbFieldName.ARRAY_NEW_SIZE, 2)));
+ // captureMode=change_streams does not look up the full document, so
an update has no `after`
+ final Struct value = new Struct(ENVELOPE_SCHEMA)
+ .put(MongoDbFieldName.UPDATE_DESCRIPTION, updateDescription)
+ .put(Envelope.FieldName.OPERATION,
Envelope.Operation.UPDATE.code())
+ .put(Envelope.FieldName.TIMESTAMP, 1234L);
+
+ final Message inMessage = endpoint.createDbzExchange(null,
createRecord(ENVELOPE_SCHEMA, value)).getIn();
+
+ assertEquals(Envelope.Operation.UPDATE.code(),
inMessage.getHeader(DebeziumConstants.HEADER_OPERATION));
+ assertNull(inMessage.getBody());
+
+ final Map<?, ?> header =
inMessage.getHeader(DebeziumConstants.HEADER_UPDATE_DESCRIPTION, Map.class);
+ assertNotNull(header);
+ assertEquals("{\"name\": \"john\"}",
header.get(MongoDbFieldName.UPDATED_FIELDS));
+ assertEquals(List.of("nickname"),
header.get(MongoDbFieldName.REMOVED_FIELDS));
+ assertEquals(List.of(Map.of(MongoDbFieldName.ARRAY_FIELD_NAME, "tags",
MongoDbFieldName.ARRAY_NEW_SIZE, 2)),
+ header.get(MongoDbFieldName.TRUNCATED_ARRAYS));
+ }
+
+ @Test
+ void testIfCreatesExchangeFromFlattenedRecord() {
+ // the document flattened by the ExtractNewDocumentState SMT
+ final Schema documentSchema =
SchemaBuilder.struct().name("dummy.inventory.customers")
+ .field("_id", Schema.INT32_SCHEMA)
+ .field("name", Schema.STRING_SCHEMA)
+ .build();
+ final Struct document = new Struct(documentSchema).put("_id",
1).put("name", "jane");
+
+ final Exchange exchange = endpoint.createDbzExchange(null,
createRecord(documentSchema, document));
+ final Message inMessage = exchange.getIn();
+
+ final Struct body = inMessage.getBody(Struct.class);
+ assertNotNull(body);
+ assertEquals("jane", body.getString("name"));
+
assertNull(inMessage.getHeader(DebeziumConstants.HEADER_UPDATE_DESCRIPTION));
+ }
+
+ private static SourceRecord createRecord(final Schema valueSchema, final
Struct value) {
+ return new SourceRecord(new HashMap<>(), new HashMap<>(), "dummy",
null, null, valueSchema, value);
+ }
+}
diff --git
a/components/camel-debezium/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
b/components/camel-debezium/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
index 34b0b811980c..ff87d1557167 100644
---
a/components/camel-debezium/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
+++
b/components/camel-debezium/camel-debezium-mysql/src/main/docs/debezium-mysql-component.adoc
@@ -91,6 +91,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-oracle/src/main/docs/debezium-oracle-component.adoc
b/components/camel-debezium/camel-debezium-oracle/src/main/docs/debezium-oracle-component.adoc
index ea3bb07fbefe..d2555e6dcf69 100644
---
a/components/camel-debezium/camel-debezium-oracle/src/main/docs/debezium-oracle-component.adoc
+++
b/components/camel-debezium/camel-debezium-oracle/src/main/docs/debezium-oracle-component.adoc
@@ -78,6 +78,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
b/components/camel-debezium/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
index 7efd8adcf172..0f7e6e9de9a6 100644
---
a/components/camel-debezium/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
+++
b/components/camel-debezium/camel-debezium-postgres/src/main/docs/debezium-postgres-component.adoc
@@ -78,6 +78,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
diff --git
a/components/camel-debezium/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
b/components/camel-debezium/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
index 96bc0187996e..43913d21386c 100644
---
a/components/camel-debezium/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
+++
b/components/camel-debezium/camel-debezium-sqlserver/src/main/docs/debezium-sqlserver-component.adoc
@@ -76,6 +76,12 @@ If the message body is not `null` (in case of tombstones),
it contains the state
- `Struct`
- `Map` (if you use the included Type Converter from `Struct` to `Map`)
+With the Debezium `ExtractNewRecordState` single message transformation
(`additionalProperties.transforms=unwrap` and
+`additionalProperties.transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState`),
the event is flattened to
+the row, and the body is that row as a `Struct`. The `CamelDebeziumOperation`
header is then set only when the
+transformation adds the operation, with
`additionalProperties.transforms.unwrap.add.fields=op` or
+`additionalProperties.transforms.unwrap.add.headers=op` (with the default `__`
prefix).
+
See below for more details.
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 661e3e3d2b8d..f00682a2b3e4 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
@@ -916,6 +916,20 @@ the same JVM.
An application that relied on the providers being de-registered on shutdown
must now remove them itself with
`Security.removeProvider("BC")` and `Security.removeProvider("BCPQC")`.
+=== camel-debezium - the row flattened by ExtractNewRecordState is the body
+
+With the Debezium `ExtractNewRecordState` single message transformation
(`ExtractNewDocumentState` for MongoDB), the
+body is now the flattened row (or document) as a `Struct`. Before, the body
was `null`, as the component only read the
+`after` field of the change event, which the transformation removes. Any
record whose value is a `Struct` without an
+`after` field is now the body, except the events of Debezium itself (schema
change, heartbeat and transaction metadata
+events), which keep a `null` body. The `CamelDebeziumOperation` header is read
from the `__op` field or header that
+the transformation adds with `add.fields=op` or `add.headers=op`.
+
+The `debezium-mongodb` consumer no longer reads the `patch` field for an
update, which Debezium removed in 2.0. The
+changed fields of an update are now in the new
`CamelDebeziumUpdateDescription` header.
+
+The `Struct` to `Map` type converter now also converts a `Struct` inside a
list or a map.
+
=== camel-jbang - camel validate normalize writes one file per input file to
an output directory
`camel validate normalize <files> --output=<dir>` now writes one normalized
file per input file into the
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
index 47858ecabd0b..4a0f0a0cecaf 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/DebeziumMongodbEndpointBuilderFactory.java
@@ -2439,6 +2439,21 @@ public interface DebeziumMongodbEndpointBuilderFactory {
public String debeziumDdlSQL() {
return "CamelDebeziumDdlSQL";
}
+ /**
+ * If presents, the description of an update: updatedFields (the
changed
+ * fields as a JSON String), removedFields (the names of the removed
+ * fields) and truncatedArrays (the arrays that were truncated, with
+ * their field and new size).
+ *
+ * The option is a: {@code Map<String, Object>} type.
+ *
+ * Group: consumer
+ *
+ * @return the name of the header {@code DebeziumUpdateDescription}.
+ */
+ public String debeziumUpdateDescription() {
+ return "CamelDebeziumUpdateDescription";
+ }
}
static DebeziumMongodbEndpointBuilder endpointBuilder(String
componentName, String path) {
class DebeziumMongodbEndpointBuilderImpl extends
AbstractEndpointBuilder implements DebeziumMongodbEndpointBuilder,
AdvancedDebeziumMongodbEndpointBuilder {
