Skip to content

Commit c623ec0

Browse files
CalvinnixCalvinnix
authored
KAFKA-165: Add change.stream.show.expanded.events property (#172)
1 parent 2855ca9 commit c623ec0

File tree

8 files changed

+262
-8
lines changed

8 files changed

+262
-8
lines changed

‎src/integrationTest/java/com/mongodb/kafka/connect/mongodb/MongoKafkaTestCase.java

+5
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,7 @@ public boolean isReplicaSetOrSharded() {
114114
private static final int FOUR_DOT_TWO_WIRE_VERSION = 8;
115115
public static final int FOUR_DOT_FOUR_WIRE_VERSION = 9;
116116
private static final int SIX_DOT_ZERO_WIRE_VERSION = 17;
117+
private static final int SEVEN_DOT_ZERO_WIRE_VERSION = 21;
117118

118119
public boolean isGreaterThanThreeDotSix() {
119120
return getMaxWireVersion() > THREE_DOT_SIX_WIRE_VERSION;
@@ -135,6 +136,10 @@ public boolean isAtLeastSixDotZero() {
135136
return getMaxWireVersion() >= SIX_DOT_ZERO_WIRE_VERSION;
136137
}
137138

139+
public boolean isAtLeastSevenDotZero() {
140+
return getMaxWireVersion() >= SEVEN_DOT_ZERO_WIRE_VERSION;
141+
}
142+
138143
public int getMaxWireVersion() {
139144
Document isMaster =
140145
MONGODB

‎src/integrationTest/java/com/mongodb/kafka/connect/source/MongoSourceTaskIntegrationTest.java

+139
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646

4747
import java.time.Instant;
4848
import java.util.ArrayList;
49+
import java.util.Arrays;
4950
import java.util.HashMap;
5051
import java.util.List;
5152
import java.util.Locale;
@@ -1053,6 +1054,144 @@ void testFullDocumentBeforeChange() {
10531054
}
10541055
}
10551056

1057+
@Test
1058+
@DisplayName("Ensure disambiguatedPaths exist when showExpandedEvents is true")
1059+
void testDisambiguatedPathsExistWhenShowExpandedEventsIsTrue() {
1060+
assumeTrue(isAtLeastSevenDotZero());
1061+
MongoDatabase db = getDatabaseWithPostfix();
1062+
try (AutoCloseableSourceTask task = createSourceTask()) {
1063+
MongoCollection<Document> coll = db.getCollection("coll");
1064+
coll.drop();
1065+
db.createCollection(coll.getNamespace().getCollectionName(), new CreateCollectionOptions());
1066+
HashMap<String, String> cfg = new HashMap<>();
1067+
cfg.put(
1068+
MongoSourceConfig.OUTPUT_FORMAT_VALUE_CONFIG,
1069+
OutputFormat.SCHEMA.name().toLowerCase(Locale.ROOT));
1070+
cfg.put(MongoSourceConfig.SHOW_EXPANDED_EVENTS_CONFIG, "true");
1071+
task.start(cfg);
1072+
int id = 0;
1073+
Document expected = new Document("_id", id);
1074+
coll.insertOne(expected);
1075+
coll.updateOne(Filters.eq(id), Document.parse("{ $set: { foo: 1 } }"));
1076+
coll.deleteOne(Filters.eq(id));
1077+
List<SourceRecord> records = getNextResults(task);
1078+
assertEquals(3, records.size());
1079+
Struct update = (Struct) records.get(1).value();
1080+
assertEquals(OperationType.UPDATE.getValue(), update.getString("operationType"));
1081+
Struct updateDescription = (Struct) update.get("updateDescription");
1082+
assertEquals("{}", updateDescription.getString("disambiguatedPaths"));
1083+
} finally {
1084+
db.drop();
1085+
}
1086+
}
1087+
1088+
@Test
1089+
@DisplayName("Ensure disambiguatedPaths don't exist when showExpandedEvents is false")
1090+
void testDisambiguatedPathsDontExistWhenShowExpandedEventsIsTrue() {
1091+
assumeTrue(isAtLeastSevenDotZero());
1092+
MongoDatabase db = getDatabaseWithPostfix();
1093+
try (AutoCloseableSourceTask task = createSourceTask()) {
1094+
MongoCollection<Document> coll = db.getCollection("coll");
1095+
coll.drop();
1096+
db.createCollection(coll.getNamespace().getCollectionName(), new CreateCollectionOptions());
1097+
HashMap<String, String> cfg = new HashMap<>();
1098+
cfg.put(
1099+
MongoSourceConfig.OUTPUT_FORMAT_VALUE_CONFIG,
1100+
OutputFormat.SCHEMA.name().toLowerCase(Locale.ROOT));
1101+
cfg.put(MongoSourceConfig.SHOW_EXPANDED_EVENTS_CONFIG, "false");
1102+
task.start(cfg);
1103+
int id = 0;
1104+
Document expected = new Document("_id", id);
1105+
coll.insertOne(expected);
1106+
coll.updateOne(Filters.eq(id), Document.parse("{ $set: { foo: 1 } }"));
1107+
coll.deleteOne(Filters.eq(id));
1108+
List<SourceRecord> records = getNextResults(task);
1109+
assertEquals(3, records.size());
1110+
Struct update = (Struct) records.get(1).value();
1111+
assertEquals(OperationType.UPDATE.getValue(), update.getString("operationType"));
1112+
Struct updateDescription = (Struct) update.get("updateDescription");
1113+
assertNull(updateDescription.getString("disambiguatedPaths"));
1114+
} finally {
1115+
db.drop();
1116+
}
1117+
}
1118+
1119+
@Test
1120+
@DisplayName("Ensure disambiguatedPaths don't exist by default")
1121+
void testDisambiguatedPathsDontExistByDefault() {
1122+
assumeTrue(isAtLeastSevenDotZero());
1123+
MongoDatabase db = getDatabaseWithPostfix();
1124+
try (AutoCloseableSourceTask task = createSourceTask()) {
1125+
MongoCollection<Document> coll = db.getCollection("coll");
1126+
coll.drop();
1127+
db.createCollection(coll.getNamespace().getCollectionName(), new CreateCollectionOptions());
1128+
HashMap<String, String> cfg = new HashMap<>();
1129+
cfg.put(
1130+
MongoSourceConfig.OUTPUT_FORMAT_VALUE_CONFIG,
1131+
OutputFormat.SCHEMA.name().toLowerCase(Locale.ROOT));
1132+
task.start(cfg);
1133+
int id = 0;
1134+
Document expected = new Document("_id", id);
1135+
coll.insertOne(expected);
1136+
coll.updateOne(Filters.eq(id), Document.parse("{ $set: { foo: 1 } }"));
1137+
coll.deleteOne(Filters.eq(id));
1138+
List<SourceRecord> records = getNextResults(task);
1139+
assertEquals(3, records.size());
1140+
Struct update = (Struct) records.get(1).value();
1141+
assertEquals(OperationType.UPDATE.getValue(), update.getString("operationType"));
1142+
Struct updateDescription = (Struct) update.get("updateDescription");
1143+
assertNull(updateDescription.getString("disambiguatedPaths"));
1144+
} finally {
1145+
db.drop();
1146+
}
1147+
}
1148+
1149+
@Test
1150+
@DisplayName("Ensure truncatedArrays works")
1151+
void testTruncatedArrays() {
1152+
assumeTrue(isAtLeastSixDotZero());
1153+
MongoDatabase db = getDatabaseWithPostfix();
1154+
try (AutoCloseableSourceTask task = createSourceTask()) {
1155+
MongoCollection<Document> coll = db.getCollection("coll");
1156+
coll.drop();
1157+
db.createCollection(coll.getNamespace().getCollectionName(), new CreateCollectionOptions());
1158+
HashMap<String, String> cfg = new HashMap<>();
1159+
cfg.put(
1160+
MongoSourceConfig.OUTPUT_FORMAT_VALUE_CONFIG,
1161+
OutputFormat.SCHEMA.name().toLowerCase(Locale.ROOT));
1162+
task.start(cfg);
1163+
int id = 0;
1164+
Document expected =
1165+
new Document("_id", id)
1166+
.append("items", Arrays.asList(2, 30, 5, 10, 11, 100, 200, 250, 300, 5, 600));
1167+
coll.insertOne(expected);
1168+
coll.updateOne(
1169+
Filters.eq(id),
1170+
singletonList(Document.parse("{ $set: { items: [2,30,5,10,11,100,200,250,300,5] } }")));
1171+
coll.deleteOne(Filters.eq(id));
1172+
List<SourceRecord> records = getNextResults(task);
1173+
assertEquals(3, records.size());
1174+
Struct update = (Struct) records.get(1).value();
1175+
assertEquals(OperationType.UPDATE.getValue(), update.getString("operationType"));
1176+
Struct updateDescription = (Struct) update.get("updateDescription");
1177+
1178+
Schema schema =
1179+
SchemaBuilder.struct()
1180+
.name("truncatedArray")
1181+
.field("field", Schema.STRING_SCHEMA)
1182+
.field("newSize", Schema.INT32_SCHEMA)
1183+
.build();
1184+
1185+
Struct truncatedArrayStruct = new Struct(schema).put("field", "items").put("newSize", 10);
1186+
1187+
List<Struct> expectedTruncatedArray = new ArrayList<>();
1188+
expectedTruncatedArray.add(truncatedArrayStruct);
1189+
assertEquals(expectedTruncatedArray, updateDescription.getArray("truncatedArrays"));
1190+
} finally {
1191+
db.drop();
1192+
}
1193+
}
1194+
10561195
/**
10571196
* We insert a document into a collection before starting the {@link MongoSourceTask}, yet we
10581197
* observe the change due to specifying {@link

‎src/main/java/com/mongodb/kafka/connect/sink/cdc/mongodb/operations/OperationHelper.java

+24-2
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,10 @@ final class OperationHelper {
3737
private static final String UPDATE_DESCRIPTION = "updateDescription";
3838
private static final String UPDATED_FIELDS = "updatedFields";
3939
private static final String REMOVED_FIELDS = "removedFields";
40+
private static final String TRUNCATED_ARRAYS = "truncatedArrays";
41+
private static final String DISAMBIGUATED_PATHS = "disambiguatedPaths";
4042
private static final Set<String> UPDATE_DESCRIPTION_FIELDS =
41-
new HashSet<>(asList(UPDATED_FIELDS, REMOVED_FIELDS));
43+
new HashSet<>(asList(UPDATED_FIELDS, REMOVED_FIELDS, TRUNCATED_ARRAYS, DISAMBIGUATED_PATHS));
4244

4345
private static final String SET = "$set";
4446
private static final String UNSET = "$unset";
@@ -125,14 +127,34 @@ static BsonDocument getUpdateDocument(final BsonDocument changeStreamDocument) {
125127
REMOVED_FIELDS, updateDescription.get(REMOVED_FIELDS), updateDescription.toJson()));
126128
}
127129

130+
if (updateDescription.containsKey(TRUNCATED_ARRAYS)
131+
&& !updateDescription.get(TRUNCATED_ARRAYS).isArray()) {
132+
throw new DataException(
133+
format(
134+
"Unexpected %s field type, expected an array but found `%s`: %s",
135+
TRUNCATED_ARRAYS,
136+
updateDescription.get(TRUNCATED_ARRAYS),
137+
updateDescription.toJson()));
138+
}
139+
140+
if (updateDescription.containsKey(DISAMBIGUATED_PATHS)
141+
&& !updateDescription.get(DISAMBIGUATED_PATHS).isDocument()) {
142+
throw new DataException(
143+
format(
144+
"Unexpected %s field type, expected an array but found `%s`: %s",
145+
DISAMBIGUATED_PATHS,
146+
updateDescription.get(DISAMBIGUATED_PATHS),
147+
updateDescription.toJson()));
148+
}
149+
128150
BsonDocument updatedFields = updateDescription.getDocument(UPDATED_FIELDS);
129151
BsonArray removedFields = updateDescription.getArray(REMOVED_FIELDS);
130152
BsonDocument unsetDocument = new BsonDocument();
131153
for (final BsonValue removedField : removedFields) {
132154
if (!removedField.isString()) {
133155
throw new DataException(
134156
format(
135-
"Unexpected value type in %s, expected an string but found `%s`: %s",
157+
"Unexpected value type in %s, expected a string but found `%s`: %s",
136158
REMOVED_FIELDS, removedField, updateDescription.toJson()));
137159
}
138160
unsetDocument.append(removedField.asString().getValue(), EMPTY_STRING);

‎src/main/java/com/mongodb/kafka/connect/source/MongoSourceConfig.java

+30
Original file line numberDiff line numberDiff line change
@@ -323,6 +323,21 @@ public class MongoSourceConfig extends AbstractConfig {
323323
+ "See https://www.mongodb.com/docs/manual/reference/method/db.collection.watch/ for more details and possible values.";
324324
private static final String FULL_DOCUMENT_DEFAULT = EMPTY_STRING;
325325

326+
public static final String SHOW_EXPANDED_EVENTS_CONFIG = "change.stream.show.expanded.events";
327+
private static final String SHOW_EXPANDED_EVENTS_DISPLAY =
328+
"The `showExpandedEvents` configuration.";
329+
private static final String SHOW_EXPANDED_EVENTS_DOC =
330+
"Determines if change streams notifies for DDL events, like the createIndexes and dropIndexes events.\n"
331+
+ "New in version 6.0.\n"
332+
+ "See https://www.mongodb.com/docs/manual/reference/change-events/#std-label-change-streams-expanded-events for more "
333+
+ "details on showExpandedEvents.\n"
334+
+ "This setting is required to show updateDescription.disambiguatedPaths in update events, "
335+
+ "helping clarify changes that involve ambiguous fields.\n"
336+
+ "New in version 6.1.\n"
337+
+ "See https://www.mongodb.com/docs/manual/reference/change-events/update/#path-disambiguation for more details on "
338+
+ "disambiguatedPaths.";
339+
private static final boolean SHOW_EXPANDED_EVENTS_DEFAULT = false;
340+
326341
public static final String COLLATION_CONFIG = "collation";
327342
private static final String COLLATION_DISPLAY = "The collation options";
328343
private static final String COLLATION_DOC =
@@ -803,6 +818,10 @@ Optional<FullDocument> getFullDocument() {
803818
}
804819
}
805820

821+
boolean getShowExpandedEvents() {
822+
return getBoolean(SHOW_EXPANDED_EVENTS_CONFIG);
823+
}
824+
806825
StartupConfig getStartupConfig() {
807826
StartupConfig result = startupConfig;
808827
if (result != null) {
@@ -1092,6 +1111,17 @@ public Map<String, ConfigValue> validateAll(final Map<String, String> props) {
10921111
FULL_DOCUMENT_DISPLAY,
10931112
Validators.EnumValidatorAndRecommender.in(FullDocument.values(), FullDocument::getValue));
10941113

1114+
configDef.define(
1115+
SHOW_EXPANDED_EVENTS_CONFIG,
1116+
Type.BOOLEAN,
1117+
SHOW_EXPANDED_EVENTS_DEFAULT,
1118+
Importance.MEDIUM,
1119+
SHOW_EXPANDED_EVENTS_DOC,
1120+
group,
1121+
++orderInGroup,
1122+
Width.MEDIUM,
1123+
SHOW_EXPANDED_EVENTS_DISPLAY);
1124+
10951125
configDef.define(
10961126
COLLATION_CONFIG,
10971127
Type.STRING,

‎src/main/java/com/mongodb/kafka/connect/source/StartedMongoSourceTask.java

+1
Original file line numberDiff line numberDiff line change
@@ -681,6 +681,7 @@ private static ChangeStreamIterable<Document> getChangeStreamIterable(
681681
if (batchSize > 0) {
682682
changeStream.batchSize(batchSize);
683683
}
684+
changeStream.showExpandedEvents(sourceConfig.getShowExpandedEvents());
684685
sourceConfig.getFullDocumentBeforeChange().ifPresent(changeStream::fullDocumentBeforeChange);
685686
sourceConfig.getFullDocument().ifPresent(changeStream::fullDocument);
686687
sourceConfig.getCollation().ifPresent(changeStream::collation);

‎src/main/java/com/mongodb/kafka/connect/source/schema/AvroSchemaDefaults.java

+9-2
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,15 @@ public final class AvroSchemaDefaults {
5151
+ " \"type\": [{\"name\": \"updateDescription\", \"type\": \"record\", \"fields\": ["
5252
+ " {\"name\": \"updatedFields\", \"type\": [\"string\", \"null\"]},"
5353
+ " {\"name\": \"removedFields\","
54-
+ " \"type\": [{\"type\": \"array\", \"items\": \"string\"}, \"null\"]"
55-
+ " }] }, \"null\"] },"
54+
+ " \"type\": [{\"type\": \"array\", \"items\": \"string\"}, \"null\"]},"
55+
+ " {\"name\": \"truncatedArrays\","
56+
+ " \"type\": [{ \"type\":\"array\", \"items\": {\"type\": \"record\","
57+
+ " \"name\": \"truncatedArray\", \"fields\": ["
58+
+ " {\"name\": \"field\", \"type\": \"string\"},"
59+
+ " {\"name\": \"newSize\", \"type\": \"int\"} ] }"
60+
+ " }, \"null\" ] },"
61+
+ " {\"name\": \"disambiguatedPaths\", \"type\": [\"string\", \"null\"]}"
62+
+ " ]}, \"null\"] },"
5663
+ " { \"name\": \"clusterTime\", \"type\": [\"string\", \"null\"] },"
5764
+ " { \"name\": \"txnNumber\", \"type\": [\"long\", \"null\"]},"
5865
+ " { \"name\": \"lsid\", \"type\": [{\"name\": \"lsid\", \"type\": \"record\","

‎src/test/java/com/mongodb/kafka/connect/sink/cdc/mongodb/operations/UpdateTest.java

+26-3
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,12 @@ class UpdateTest {
5252
+ " updatedFields: {"
5353
+ " email: '[email protected]'"
5454
+ " },"
55-
+ " removedFields: ['phoneNumber']"
55+
+ " removedFields: ['phoneNumber'],"
56+
+ " truncatedArrays: [{ field: 'foo', newSize: 1 }],"
57+
+ " disambiguatedPaths: {"
58+
+ " 'home.town': [ 'home.town' ],"
59+
+ " 'residences.0.0': [ 'residences', 0, '0' ]"
60+
+ " }"
5661
+ " },"
5762
+ " fullDocument: {"
5863
+ " _id: ObjectId(\"58a4eb4a30c75625e00d2820\"),"
@@ -148,7 +153,7 @@ void testMissingChangeEventData() {
148153
new SinkDocument(
149154
null,
150155
BsonDocument.parse(
151-
"{documentKey: {}, updateDescription: {updatedFields: 1}}")))),
156+
"{documentKey: {}, updateDescription: {updatedFields: 1, removedFields: []}}")))),
152157
() ->
153158
assertThrows(
154159
DataException.class,
@@ -157,7 +162,25 @@ void testMissingChangeEventData() {
157162
new SinkDocument(
158163
null,
159164
BsonDocument.parse(
160-
"{documentKey: {}, updateDescription: {removedFields: 1}}")))),
165+
"{documentKey: {}, updateDescription: {updatedFields: {}, removedFields: 1}}")))),
166+
() ->
167+
assertThrows(
168+
DataException.class,
169+
() ->
170+
UPDATE.perform(
171+
new SinkDocument(
172+
null,
173+
BsonDocument.parse(
174+
"{documentKey: {}, updateDescription: {updatedFields: {}, removedFields: [], truncatedArrays: 1}}")))),
175+
() ->
176+
assertThrows(
177+
DataException.class,
178+
() ->
179+
UPDATE.perform(
180+
new SinkDocument(
181+
null,
182+
BsonDocument.parse(
183+
"{documentKey: {}, updateDescription: {updatedFields: {}, removedFields: [], disambiguatedPaths: 1}}")))),
161184
() ->
162185
assertThrows(
163186
DataException.class,

0 commit comments

Comments
 (0)