ExtractNewDocumentState implements Transformation<R> { @Override public R apply(R record) { if (!smtManager.isValidKey(record)) { return record; } final R keyRecord = keyExtractor.apply(record); BsonDocument keyDocument = BsonDocument.parse("{ \"id\" : " + keyRecord.key().toString() + "}"); BsonDocument valueDocument = new BsonDocument(); if (record.value() == null) { if (dropTombstones) { LOGGER.trace("Tombstone {} arrived and requested to be dropped", record.key()); return null; } if (!additionalHeaders.isEmpty()) { Headers headersToAdd = makeHeaders(additionalHeaders, (Struct) record.value()); headersToAdd.forEach(h -> record.headers().add(h)); } else if (addOperationHeader) { LOGGER.warn("operation.header has been deprecated and is scheduled for removal. Use add.headers instead."); record.headers().addString(ExtractNewRecordStateConfigDefinition.DEBEZIUM_OPERATION_HEADER_KEY, Operation.DELETE.code()); } return newRecord(record, keyDocument, valueDocument); } if (!smtManager.isValidEnvelope(record)) { return record; } final R afterRecord = afterExtractor.apply(record); final R patchRecord = patchExtractor.apply(record); if (!additionalHeaders.isEmpty()) { Headers headersToAdd = makeHeaders(additionalHeaders, (Struct) record.value()); headersToAdd.forEach(h -> record.headers().add(h)); } else if (addOperationHeader) { LOGGER.warn("operation.header has been deprecated and is scheduled for removal. Use add.headers instead."); record.headers().addString(ExtractNewRecordStateConfigDefinition.DEBEZIUM_OPERATION_HEADER_KEY, ((Struct) record.value()).get("op").toString()); } if (afterRecord.value() != null) { valueDocument = getInsertDocument(afterRecord, keyDocument); } if (afterRecord.value() == null && patchRecord.value() != null) { valueDocument = getUpdateDocument(patchRecord, keyDocument); } boolean isDeletion = false; if (afterRecord.value() == null && patchRecord.value() == null) { if (handleDeletes.equals(DeleteHandling.DROP)) { LOGGER.trace("Delete {} arrived and requested to be dropped", record.key()); return null; } isDeletion = true; } if (handleDeletes.equals(DeleteHandling.REWRITE)) { valueDocument.append(ExtractNewRecordStateConfigDefinition.DELETED_FIELD, new BsonBoolean(isDeletion)); } return newRecord(record, keyDocument, valueDocument); }  @Override R apply(R record); @Override ConfigDef config(); @Override void close(); @Override void configure(final Map<String, ?> map); static final Field SANITIZE_FIELD_NAMES; static final Field ADD_SOURCE_FIELDS; static final Field OPERATION_HEADER; }
@Test @FixFor("DBZ-1430") public void shouldPassHeartbeatMessages() { Schema valueSchema = SchemaBuilder.struct() .name("io.debezium.connector.common.Heartbeat") .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema).put(AbstractSourceInfo.TIMESTAMP_KEY, 1565787098802L); Schema keySchema = SchemaBuilder.struct() .name("io.debezium.connector.common.ServerNameKey") .field("serverName", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("serverName", "op.with.heartbeat"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, valueSchema, value); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isSameAs(eventRecord); }
@Test @FixFor("DBZ-1430") public void shouldSkipMessagesWithoutDebeziumCdcEnvelopeDueToMissingSchemaName() { Schema valueSchema = SchemaBuilder.struct() .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema); Schema keySchema = SchemaBuilder.struct() .name("op.with.heartbeat.Key") .field("id", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("id", "123"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, valueSchema, value); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isSameAs(eventRecord); }
@Test @FixFor("DBZ-1430") public void shouldSkipMessagesWithoutDebeziumCdcEnvelopeDueToMissingSchemaNameSuffix() { Schema valueSchema = SchemaBuilder.struct() .name("io.debezium.connector.common.Heartbeat") .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema); Schema keySchema = SchemaBuilder.struct() .name("op.with.heartbeat.Key") .field("id", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("id", "123"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, valueSchema, value); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isSameAs(eventRecord); }
@Test @FixFor("DBZ-1430") public void shouldSkipMessagesWithoutDebeziumCdcEnvelopeDueToMissingValueSchema() { Schema valueSchema = SchemaBuilder.struct() .name("io.debezium.connector.common.Heartbeat.Envelope") .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema); Schema keySchema = SchemaBuilder.struct() .name("op.with.heartbeat.Key") .field("id", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("id", "123"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, null, value); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isSameAs(eventRecord); }
@Test @FixFor("DBZ-1430") @SkipWhenKafkaVersion(check = GREATER_THAN_OR_EQUAL, value = KAFKA_241, description = "Kafka throws IllegalArgumentException after 2.4.1") public void shouldFailWhenTheSchemaLooksValidButDoesNotHaveTheCorrectFieldsPreKafka241() { Schema valueSchema = SchemaBuilder.struct() .name("io.debezium.connector.common.Heartbeat.Envelope") .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema); Schema keySchema = SchemaBuilder.struct() .name("op.with.heartbeat.Key") .field("id", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("id", "123"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, valueSchema, value); exceptionRule.expect(NullPointerException.class); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isNull(); }
@Test @FixFor("DBZ-1430") @SkipWhenKafkaVersion(check = LESS_THAN, value = KAFKA_241, description = "Kafka throws NullPointerException prior to 2.4.1") public void shouldFailWhenTheSchemaLooksValidButDoesNotHaveTheCorrectFieldsPostKafka241() { Schema valueSchema = SchemaBuilder.struct() .name("io.debezium.connector.common.Heartbeat.Envelope") .field(AbstractSourceInfo.TIMESTAMP_KEY, Schema.INT64_SCHEMA) .build(); Struct value = new Struct(valueSchema); Schema keySchema = SchemaBuilder.struct() .name("op.with.heartbeat.Key") .field("id", Schema.STRING_SCHEMA) .build(); Struct key = new Struct(keySchema).put("id", "123"); final SourceRecord eventRecord = new SourceRecord( new HashMap<>(), new HashMap<>(), "op.with.heartbeat", keySchema, key, valueSchema, value); exceptionRule.expect(IllegalArgumentException.class); SourceRecord transformed = transformation.apply(eventRecord); assertThat(transformed).isNull(); }