diff --git a/kafka-connect-avro-converter-shaded/pom.xml b/kafka-connect-avro-converter-shaded/pom.xml new file mode 100644 index 0000000000000..100636771286f --- /dev/null +++ b/kafka-connect-avro-converter-shaded/pom.xml @@ -0,0 +1,128 @@ + + + + 4.0.0 + + pulsar + org.apache.pulsar + 2.6.0-SNAPSHOT + .. + + + kafka-connect-avro-converter-shaded + Apache Pulsar :: Kafka Connect Avro Converter shaded + + + + + io.confluent + kafka-connect-avro-converter + ${confluent.version} + + + org.codehaus.jackson + jackson-core-asl + ${kafka-avro-convert-jackson.version} + + + org.codehaus.jackson + jackson-mapper-asl + ${kafka-avro-convert-jackson.version} + + + + + + + org.apache.maven.plugins + maven-shade-plugin + + + package + + shade + + + + + true + true + + + + io.confluent:* + io.confluent:kafka-avro-serializer + io.confluent:kafka-schema-registry-client + io.confluent:common-config + io.confluent:common-utils + org.apache.avro:* + + org.codehaus.jackson:jackson-core-asl + org.codehaus.jackson:jackson-mapper-asl + com.thoughtworks.paranamer:paranamer + org.xerial.snappy:snappy-java + org.apache.commons:commons-compress + org.tukaani:xz + + + + + io.confluent + org.apache.pulsar.kafka.shade.io.confluent + + + org.apache.avro + org.apache.pulsar.kafka.shade.avro + + + org.codehaus.jackson + org.apache.pulsar.kafka.shade.org.codehaus.jackson + + + com.thoughtworks.paranamer + org.apache.pulsar.kafka.shade.com.thoughtworks.paranamer + + + org.xerial.snappy + org.apache.pulsar.kafka.shade.org.xerial.snappy + + + org.apache.commons + org.apache.pulsar.kafka.shade.org.apache.commons + + + org.tukaani + org.apache.pulsar.kafka.shade.org.tukaani + + + + + + + + + + + + \ No newline at end of file diff --git a/pom.xml b/pom.xml index 471abc248f27a..ddeda8fc081da 100644 --- a/pom.xml +++ b/pom.xml @@ -117,6 +117,9 @@ flexible messaging model and an intuitive client API. pulsar-io + + kafka-connect-avro-converter-shaded + examples @@ -208,6 +211,8 @@ flexible messaging model and an intuitive client API. 25.1-jre 1.0 0.12.0 + 5.3.2 + 1.9.13 3.6.0 @@ -220,6 +225,7 @@ flexible messaging model and an intuitive client API. 3.25.0-GA 2.3.1 1.5.0 + 5.2.2 0.6.1 @@ -1715,5 +1721,9 @@ flexible messaging model and an intuitive client API. spring-plugins-release https://repo.spring.io/plugins-release/ + + confluent + http://packages.confluent.io/maven/ + diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java index f9faa0b1156bc..fc0608bbb2ff6 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/StructSchema.java @@ -41,6 +41,7 @@ import org.apache.pulsar.client.api.schema.SchemaInfoProvider; import org.apache.pulsar.client.api.schema.SchemaReader; import org.apache.pulsar.client.api.schema.SchemaWriter; +import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema; import org.apache.pulsar.common.protocol.schema.BytesSchemaVersion; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.common.schema.SchemaType; @@ -78,6 +79,11 @@ public SchemaReader load(BytesSchemaVersion schemaVersion) { protected StructSchema(SchemaInfo schemaInfo) { this.schema = parseAvroSchema(new String(schemaInfo.getSchema(), UTF_8)); this.schemaInfo = schemaInfo; + + if (schemaInfo.getProperties().containsKey(GenericAvroSchema.OFFSET_PROP)) { + this.schema.addProp(GenericAvroSchema.OFFSET_PROP, + schemaInfo.getProperties().get(GenericAvroSchema.OFFSET_PROP)); + } } public org.apache.avro.Schema getAvroSchema() { diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReader.java index 2ac0c2f2695bf..22c63a94a5f18 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReader.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReader.java @@ -47,6 +47,8 @@ public class GenericAvroReader implements SchemaReader { private final List fields; private final Schema schema; private final byte[] schemaVersion; + private int offset; + public GenericAvroReader(Schema schema) { this(null, schema, null); } @@ -65,12 +67,22 @@ public GenericAvroReader(Schema writerSchema, Schema readerSchema, byte[] schema } this.byteArrayOutputStream = new ByteArrayOutputStream(); this.encoder = EncoderFactory.get().binaryEncoder(this.byteArrayOutputStream, encoder); + + if (schema.getObjectProp(GenericAvroSchema.OFFSET_PROP) != null) { + this.offset = Integer.parseInt(schema.getObjectProp(GenericAvroSchema.OFFSET_PROP).toString()); + } else { + this.offset = 0; + } + } @Override public GenericAvroRecord read(byte[] bytes, int offset, int length) { try { - Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length, null); + if (offset == 0 && this.offset > 0) { + offset = this.offset; + } + Decoder decoder = DecoderFactory.get().binaryDecoder(bytes, offset, length - offset, null); org.apache.avro.generic.GenericRecord avroRecord = (org.apache.avro.generic.GenericRecord)reader.read( null, @@ -101,5 +113,9 @@ public GenericRecord read(InputStream inputStream) { } } + public int getOffset() { + return offset; + } + private static final Logger log = LoggerFactory.getLogger(GenericAvroReader.class); } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java index 98e646ea2ba50..94c5ba1f9911c 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroSchema.java @@ -33,6 +33,8 @@ @Slf4j public class GenericAvroSchema extends GenericSchemaImpl { + public final static String OFFSET_PROP = "__AVRO_READ_OFFSET__"; + public GenericAvroSchema(SchemaInfo schemaInfo) { this(schemaInfo, true); } @@ -73,6 +75,8 @@ protected SchemaReader loadReader(BytesSchemaVersion schemaVersio schemaInfo); Schema writerSchema = parseAvroSchema(schemaInfo.getSchemaDefinition()); Schema readerSchema = useProvidedSchemaAsReaderSchema ? schema : writerSchema; + readerSchema.addProp(OFFSET_PROP, schemaInfo.getProperties().getOrDefault(OFFSET_PROP, "0")); + return new GenericAvroReader( writerSchema, readerSchema, diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReaderTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReaderTest.java index d77d0f2995414..15717947a22d9 100644 --- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReaderTest.java +++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/generic/GenericAvroReaderTest.java @@ -20,6 +20,8 @@ import static org.testng.Assert.assertEquals; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; import lombok.extern.slf4j.Slf4j; import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.api.schema.SchemaDefinition; @@ -38,11 +40,12 @@ public class GenericAvroReaderTest { private AvroSchema fooSchemaNotNull; private AvroSchema fooSchema; private AvroSchema fooV2Schema; - + private AvroSchema fooOffsetSchema; @BeforeMethod public void setup() { fooSchema = AvroSchema.of(Foo.class); + fooV2Schema = AvroSchema.of(FooV2.class); fooSchemaNotNull = AvroSchema.of(SchemaDefinition .builder() @@ -50,6 +53,9 @@ public void setup() { .withPojo(Foo.class) .build()); + fooOffsetSchema = AvroSchema.of(Foo.class); + fooOffsetSchema.getAvroSchema().addProp(GenericAvroSchema.OFFSET_PROP, 5); + foo = new Foo(); foo.setField1("foo1"); foo.setField2("bar1"); @@ -83,4 +89,20 @@ public void testGenericAvroReaderByReaderSchema() { assertEquals(genericRecordByReaderSchema.getField("field3"), 10); } + @Test + public void testOffsetSchema() { + byte[] fooBytes = fooOffsetSchema.encode(foo); + ByteBuf byteBuf = Unpooled.buffer(); + byteBuf.writeByte(0); + byteBuf.writeInt(10); + byteBuf.writeBytes(fooBytes); + + GenericAvroReader reader = new GenericAvroReader(fooOffsetSchema.getAvroSchema()); + assertEquals(reader.getOffset(), 5); + GenericRecord record = reader.read(byteBuf.array()); + assertEquals(record.getField("field1"), "foo1"); + assertEquals(record.getField("field2"), "bar1"); + assertEquals(record.getField("fieldUnableNull"), "notNull"); + } + } diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java new file mode 100644 index 0000000000000..812e8c698ec26 --- /dev/null +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/KVRecord.java @@ -0,0 +1,35 @@ +/** + * 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.pulsar.functions.api; + +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.schema.KeyValueEncodingType; + +/** + * key value schema record. + */ +public interface KVRecord extends Record { + + Schema getKeySchema(); + + Schema getValueSchema(); + + KeyValueEncodingType getKeyValueEncodingType(); + +} diff --git a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java index 3c3d7e8d78efe..9280cd71f1874 100644 --- a/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java +++ b/pulsar-functions/api-java/src/main/java/org/apache/pulsar/functions/api/Record.java @@ -19,6 +19,7 @@ package org.apache.pulsar.functions.api; import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.Schema; import java.util.Collections; import java.util.Map; @@ -43,6 +44,10 @@ default Optional getKey() { return Optional.empty(); } + default Schema getSchema() { + return null; + } + /** * Retrieves the actual data of the record. * diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java index 3e1d5c01a2d88..71a398484d6ab 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/SinkRecord.java @@ -24,8 +24,13 @@ import lombok.AllArgsConstructor; import lombok.Data; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.schema.KeyValueSchema; +import org.apache.pulsar.functions.api.KVRecord; import org.apache.pulsar.functions.api.Record; +@Slf4j @Data @AllArgsConstructor public class SinkRecord implements Record { @@ -81,4 +86,24 @@ public void fail() { public Optional getDestinationTopic() { return sourceRecord.getDestinationTopic(); } + + @Override + public Schema getSchema() { + if (sourceRecord == null) { + return null; + } + + if (sourceRecord.getSchema() != null) { + return sourceRecord.getSchema(); + } + + if (sourceRecord instanceof KVRecord) { + KVRecord kvRecord = (KVRecord) sourceRecord; + return KeyValueSchema.of(kvRecord.getKeySchema(), kvRecord.getValueSchema(), + kvRecord.getKeyValueEncodingType()); + } + + return null; + } + } diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java index 43100e46a3759..8cb67efa47b1d 100644 --- a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java +++ b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java @@ -31,7 +31,9 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TypedMessageBuilder; +import org.apache.pulsar.client.impl.schema.KeyValueSchema; import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.functions.instance.FunctionResultRouter; import org.apache.pulsar.functions.instance.SinkRecord; @@ -104,18 +106,18 @@ public Producer createProducer(PulsarClient client, String topic, String prod return builder.properties(properties).create(); } - protected Producer getProducer(String destinationTopic) { - return getProducer(destinationTopic, null, destinationTopic); + protected Producer getProducer(String destinationTopic, Schema schema) { + return getProducer(destinationTopic, null, destinationTopic, schema); } - protected Producer getProducer(String producerId, String producerName, String topicName) { + protected Producer getProducer(String producerId, String producerName, String topicName, Schema schema) { return publishProducers.computeIfAbsent(producerId, s -> { try { return createProducer( client, topicName, producerName, - schema); + schema != null ? schema : this.schema); } catch (PulsarClientException e) { log.error("Failed to create Producer while doing user publish", e); throw new RuntimeException(e); @@ -177,7 +179,17 @@ public PulsarSinkAtMostOnceProcessor(Schema schema) { @Override public TypedMessageBuilder newMessage(Record record) { - return getProducer(record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic())).newMessage(); + if (record.getSchema() != null) { + return getProducer(record + .getDestinationTopic() + .orElse(pulsarSinkConfig.getTopic()), record.getSchema()) + .newMessage(record.getSchema()); + } else { + return getProducer(record + .getDestinationTopic() + .orElse(pulsarSinkConfig.getTopic()), record.getSchema()) + .newMessage(); + } } @Override @@ -215,11 +227,17 @@ public TypedMessageBuilder newMessage(Record record) { throw new RuntimeException("PartitionId needs to be specified for every record while in Effectively-once mode"); } - return getProducer( + Producer producer = getProducer( String.format("%s-%s",record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic()), record.getPartitionId().get()), record.getPartitionId().get(), - record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic()) - ).newMessage(); + record.getDestinationTopic().orElse(pulsarSinkConfig.getTopic()), + record.getSchema() + ); + if (record.getSchema() != null) { + return producer.newMessage(record.getSchema()); + } else { + return producer.newMessage(); + } } @Override @@ -274,7 +292,9 @@ public void open(Map config, SinkContext sinkContext) throws Exc @Override public void write(Record record) { TypedMessageBuilder msg = pulsarSinkProcessor.newMessage(record); - if (record.getKey().isPresent()) { + + if (record.getKey().isPresent() && !(record.getSchema() instanceof KeyValueSchema && + ((KeyValueSchema) record.getSchema()).getKeyValueEncodingType() == KeyValueEncodingType.SEPARATED)) { msg.key(record.getKey().get()); } diff --git a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java index 244a757dbc480..b9ac10eea6918 100644 --- a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java +++ b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java @@ -261,7 +261,7 @@ public void start(boolean blocking) throws Exception { } else { File file = new File(userCodeFile); if (!file.exists()) { - throw new RuntimeException("Source archive does not exist"); + throw new RuntimeException("Source archive (" + userCodeFile + ") does not exist"); } functionDetails = SourceConfigUtils.convert(sourceConfig, SourceConfigUtils.validate(sourceConfig, null, file)); } diff --git a/pulsar-io/kafka-connect-adaptor/pom.xml b/pulsar-io/kafka-connect-adaptor/pom.xml index 854489b473637..5609cd129bf91 100644 --- a/pulsar-io/kafka-connect-adaptor/pom.xml +++ b/pulsar-io/kafka-connect-adaptor/pom.xml @@ -68,6 +68,22 @@ ${project.version} + + org.apache.pulsar + kafka-connect-avro-converter-shaded + ${project.version} + + + io.confluent + * + + + org.apache.avro + * + + + + ${project.groupId} pulsar-broker diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java index ed059bfa794fc..a178b4bcc00d3 100644 --- a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java +++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java @@ -34,6 +34,14 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; + +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import org.apache.kafka.connect.json.JsonConverterConfig; +import org.apache.pulsar.common.schema.KeyValueEncodingType; +import org.apache.pulsar.functions.api.KVRecord; +import org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter; +import org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroData; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.connect.runtime.TaskConfig; @@ -45,10 +53,14 @@ import org.apache.kafka.connect.storage.OffsetStorageReader; import org.apache.kafka.connect.storage.OffsetStorageReaderImpl; import org.apache.kafka.connect.storage.OffsetStorageWriter; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.functions.api.Record; import org.apache.pulsar.io.core.Source; import org.apache.pulsar.io.core.SourceContext; +import org.apache.pulsar.io.kafka.connect.schema.KafkaSchemaWrappedSchema; +import org.apache.pulsar.kafka.shade.io.confluent.kafka.schemaregistry.client.MockSchemaRegistryClient; +import org.apache.pulsar.kafka.shade.io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig; /** * A pulsar source that runs @@ -74,6 +86,13 @@ public class KafkaConnectSource implements Source> { // number of outstandingRecords that have been polled but not been acked private AtomicInteger outstandingRecords = new AtomicInteger(0); + private boolean jsonWithEnvelope = false; + static private final String JSON_WITH_ENVELOPE_CONFIG = "json-with-envelope"; + + private final Cache readerCache = + CacheBuilder.newBuilder().maximumSize(10000) + .expireAfterAccess(30, TimeUnit.MINUTES).build(); + @Override public void open(Map config, SourceContext sourceContext) throws Exception { Map stringConfig = new HashMap<>(); @@ -83,6 +102,14 @@ public void open(Map config, SourceContext sourceContext) throws } }); + if (config.get(JSON_WITH_ENVELOPE_CONFIG) != null) { + jsonWithEnvelope = Boolean.parseBoolean(config.get(JSON_WITH_ENVELOPE_CONFIG).toString()); + config.put(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, jsonWithEnvelope); + } else { + config.put(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, false); + } + log.info("jsonWithEnvelope: {}", jsonWithEnvelope); + // get the source class name from config and create source task from reflection sourceTask = ((Class)Class.forName(stringConfig.get(TaskConfig.TASK_CLASS_CONFIG))) .asSubclass(SourceTask.class) @@ -101,6 +128,14 @@ public void open(Map config, SourceContext sourceContext) throws .getDeclaredConstructor() .newInstance(); + if (keyConverter instanceof AvroConverter) { + keyConverter = new AvroConverter(new MockSchemaRegistryClient()); + config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock"); + } + if (valueConverter instanceof AvroConverter) { + valueConverter = new AvroConverter(new MockSchemaRegistryClient()); + config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock"); + } keyConverter.configure(config, true); valueConverter.configure(config, false); @@ -161,7 +196,9 @@ public synchronized Record> read() throws Exception { @Override public void close() { - sourceTask.stop(); + if (sourceTask != null) { + sourceTask.stop(); + } } private synchronized Record> processSourceRecord(final SourceRecord srcRecord) { @@ -174,7 +211,7 @@ private synchronized Record> processSourceRecord(final private static Optional RECORD_SEQUENCE = Optional.empty(); private static long FLUSH_TIMEOUT_MS = 2000; - private class KafkaSourceRecord implements Record> { + private class KafkaSourceRecord implements KVRecord { @Getter Optional key; @Getter @@ -188,15 +225,42 @@ private class KafkaSourceRecord implements Record> { @Getter Optional destinationTopic; + KafkaSchemaWrappedSchema keySchema; + + KafkaSchemaWrappedSchema valueSchema; + KafkaSourceRecord(SourceRecord srcRecord) { + AvroData avroData = new AvroData(1000); byte[] keyBytes = keyConverter.fromConnectData( - srcRecord.topic(), srcRecord.keySchema(), srcRecord.key()); - byte[] valueBytes = valueConverter.fromConnectData( - srcRecord.topic(), srcRecord.valueSchema(), srcRecord.value()); + srcRecord.topic(), srcRecord.keySchema(), srcRecord.key()); this.key = keyBytes != null ? Optional.of(Base64.getEncoder().encodeToString(keyBytes)) : Optional.empty(); - this.value = new KeyValue(keyBytes, valueBytes); + + byte[] valueBytes = valueConverter.fromConnectData( + srcRecord.topic(), srcRecord.valueSchema(), srcRecord.value()); + + this.value = new KeyValue<>(keyBytes, valueBytes); this.topicName = Optional.of(srcRecord.topic()); + + if (srcRecord.keySchema() != null) { + keySchema = readerCache.getIfPresent(srcRecord.keySchema()); + } + if (srcRecord.valueSchema() != null) { + valueSchema = readerCache.getIfPresent(srcRecord.valueSchema()); + } + + if (srcRecord.keySchema() != null && keySchema == null) { + keySchema = new KafkaSchemaWrappedSchema( + avroData.fromConnectSchema(srcRecord.keySchema()), keyConverter); + readerCache.put(srcRecord.keySchema(), keySchema); + } + + if (srcRecord.valueSchema() != null && valueSchema == null) { + valueSchema = new KafkaSchemaWrappedSchema( + avroData.fromConnectSchema(srcRecord.valueSchema()), valueConverter); + readerCache.put(srcRecord.valueSchema(), valueSchema); + } + this.eventTime = Optional.ofNullable(srcRecord.timestamp()); this.partitionId = Optional.of(srcRecord.sourcePartition() .entrySet() @@ -206,6 +270,38 @@ private class KafkaSourceRecord implements Record> { this.destinationTopic = Optional.of(topicNamespace + "/" + srcRecord.topic()); } + @Override + public Schema getKeySchema() { + if (jsonWithEnvelope || keySchema == null) { + return Schema.BYTES; + } else { + return keySchema; + } + } + + @Override + public Schema getValueSchema() { + if (jsonWithEnvelope || valueSchema == null) { + return Schema.BYTES; + } else { + return valueSchema; + } + } + + @Override + public KeyValueEncodingType getKeyValueEncodingType() { + if (jsonWithEnvelope) { + return KeyValueEncodingType.INLINE; + } else { + return KeyValueEncodingType.SEPARATED; + } + } + + @Override + public Schema getSchema() { + return null; + } + @Override public Optional getRecordSequence() { return RECORD_SEQUENCE; @@ -275,4 +371,5 @@ public void fail() { } } } + } diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/KafkaSchemaWrappedSchema.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/KafkaSchemaWrappedSchema.java new file mode 100644 index 0000000000000..2db9d6cd93bc6 --- /dev/null +++ b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/schema/KafkaSchemaWrappedSchema.java @@ -0,0 +1,69 @@ +/** + * 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.pulsar.io.kafka.connect.schema; + +import static java.nio.charset.StandardCharsets.UTF_8; + +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.connect.json.JsonConverter; +import org.apache.kafka.connect.storage.Converter; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.schema.generic.GenericAvroSchema; +import org.apache.pulsar.common.schema.SchemaInfo; +import org.apache.pulsar.common.schema.SchemaType; + +/** + * Wrapped schema for kafka connect schema. + */ +@Slf4j +public class KafkaSchemaWrappedSchema implements Schema, Serializable { + + private SchemaInfo schemaInfo = null; + + public KafkaSchemaWrappedSchema(org.apache.pulsar.kafka.shade.avro.Schema schema, + Converter converter) { + Map props = new HashMap<>(); + boolean isJsonConverter = converter instanceof JsonConverter; + props.put(GenericAvroSchema.OFFSET_PROP, isJsonConverter ? "0" : "5"); + this.schemaInfo = SchemaInfo.builder() + .name(isJsonConverter? "KafKaJson" : "KafkaAvro") + .type(isJsonConverter ? SchemaType.JSON : SchemaType.AVRO) + .schema(schema.toString().getBytes(UTF_8)) + .properties(props) + .build(); + } + + @Override + public byte[] encode(byte[] data) { + return data; + } + + @Override + public SchemaInfo getSchemaInfo() { + return schemaInfo; + } + + @Override + public Schema clone() { + return null; + } +} diff --git a/site2/docs/io-debezium-source.md b/site2/docs/io-debezium-source.md index bfa282ca1dbc3..360a5bf81496f 100644 --- a/site2/docs/io-debezium-source.md +++ b/site2/docs/io-debezium-source.md @@ -28,6 +28,24 @@ The configuration of Debezium source connector has the following properties. | `database.history.pulsar.service.url` | true | null | Pulsar cluster service URL for history topic. | | `pulsar.service.url` | true | null | Pulsar cluster service URL. | | `offset.storage.topic` | true | null | Record the last committed offsets that the connector successfully completes. | +| `json-with-envelope` | false | false | Present the message only consist of payload. + +### Converter Options + +1. org.apache.kafka.connect.json.JsonConverter + +This config `json-with-envelope` is valid only for the JsonConverter. It's default value is false, the consumer use the schema ` +Schema.KeyValue(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED)`, +and the message only consist of payload. + +If the config `json-with-envelope` value is true, the consumer use the schema +`Schema.KeyValue(Schema.BYTES, Schema.BYTES`, the message consist of schema and payload. + +2. org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter + +If users select the AvroConverter, then the pulsar consumer should use the schema `Schema.KeyValue(Schema.AUTO_CONSUME(), +Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED)`, and the message consist of payload. + ### MongoDB Configuration | Name | Required | Default | Description | |------|----------|---------|-------------| diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java index 730584953d431..a66f635e8d8f2 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/ChaosContainer.java @@ -29,7 +29,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang.StringUtils; import org.apache.pulsar.tests.integration.docker.ContainerExecResult; import org.apache.pulsar.tests.integration.utils.DockerUtils; import org.testcontainers.containers.GenericContainer; diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java index 70e9f451ade40..07262eec4f384 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java @@ -33,6 +33,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.impl.PulsarClientImpl; import org.apache.pulsar.client.impl.schema.AvroSchema; @@ -42,8 +43,10 @@ import org.apache.pulsar.common.policies.data.FunctionStatus; import org.apache.pulsar.common.policies.data.SinkStatus; import org.apache.pulsar.common.policies.data.SourceStatus; +import org.apache.pulsar.common.policies.data.TenantInfo; import org.apache.pulsar.common.policies.data.TopicStats; import org.apache.pulsar.common.schema.KeyValue; +import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaInfo; import org.apache.pulsar.functions.api.examples.AutoSchemaFunction; import org.apache.pulsar.functions.api.examples.AvroSchemaTestFunction; @@ -63,15 +66,18 @@ import org.apache.pulsar.tests.integration.utils.DockerUtils; import org.assertj.core.api.Assertions; import org.testcontainers.containers.GenericContainer; +import org.testcontainers.shaded.com.google.common.collect.Sets; import org.testng.annotations.Test; import org.testng.collections.Maps; import java.io.BufferedReader; import java.io.InputStreamReader; import java.time.Duration; +import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; @@ -81,6 +87,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.assertj.core.api.Assertions.assertThat; import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @@ -140,18 +147,24 @@ public void testRabbitMQSink() throws Exception { } @Test(groups = "source") - public void testDebeziumMySqlSource() throws Exception { - testDebeziumMySqlConnect(); + public void testDebeziumMySqlSourceJson() throws Exception { + testDebeziumMySqlConnect("org.apache.kafka.connect.json.JsonConverter", true); + } + + @Test(groups = "source") + public void testDebeziumMySqlSourceAvro() throws Exception { + testDebeziumMySqlConnect( + "org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter", false); } @Test(groups = "source") public void testDebeziumPostgreSqlSource() throws Exception { - testDebeziumPostgreSqlConnect(); + testDebeziumPostgreSqlConnect("org.apache.kafka.connect.json.JsonConverter", true); } @Test(groups = "source") public void testDebeziumMongoDbSource() throws Exception{ - testDebeziumMongoDbConnect(); + testDebeziumMongoDbConnect("org.apache.kafka.connect.json.JsonConverter", true); } private void testSink(SinkTester tester, boolean builtin) throws Exception { @@ -2262,15 +2275,17 @@ public void testAvroSchemaFunction() throws Exception { getFunctionInfoNotFound(functionName); } - private void testDebeziumMySqlConnect() - throws Exception { + private void testDebeziumMySqlConnect(String converterClassName, boolean jsonWithEnvelope) throws Exception { final String tenant = TopicName.PUBLIC_TENANT; final String namespace = TopicName.DEFAULT_NAMESPACE; final String outputTopicName = "debe-output-topic-name"; - final String consumeTopicName = "public/default/dbserver1.inventory.products"; - final String sourceName = "test-source-connector-" - + functionRuntimeType + "-name-" + randomName(8); + boolean isJsonConverter = converterClassName.endsWith("JsonConverter"); + final String consumeTopicName = "debezium/mysql-" + + (isJsonConverter ? "json" : "avro") + + "/dbserver1.inventory.products"; + final String sourceName = "test-source-debezium-mysql" + (isJsonConverter ? "json" : "avro") + + "-" + functionRuntimeType + "-" + randomName(8); // This is the binlog count that contained in mysql container. final int numMessages = 47; @@ -2287,28 +2302,21 @@ private void testDebeziumMySqlConnect() @Cleanup PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build(); + initNamespace(admin); + try { - // If topic already exists, we should delete it so as not to affect the following tests. - admin.topics().getStats(consumeTopicName); - admin.topics().delete(consumeTopicName); - admin.schemas().deleteSchema(consumeTopicName); - } catch (PulsarAdminException e) { - // Expected results, ignoring the exception - log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}", + SchemaInfo lastSchemaInfo = admin.schemas().getSchemaInfo(consumeTopicName); + log.info("lastSchemaInfo: {}", lastSchemaInfo == null ? "null" : lastSchemaInfo.toString()); + } catch (Exception e) { + log.warn("failed to get schemaInfo for topic: {}, exceptions message: {}", consumeTopicName, e.getMessage()); } - admin.topics().createNonPartitionedTopic(consumeTopicName); - admin.topics().createNonPartitionedTopic(outputTopicName); - @Cleanup - Consumer> consumer = client.newConsumer(KeyValueSchema.kvBytes()) - .topic(consumeTopicName) - .subscriptionName("debezium-source-tester") - .subscriptionType(SubscriptionType.Exclusive) - .subscribe(); + admin.topics().createNonPartitionedTopic(outputTopicName); @Cleanup - DebeziumMySqlSourceTester sourceTester = new DebeziumMySqlSourceTester(pulsarCluster); + DebeziumMySqlSourceTester sourceTester = new DebeziumMySqlSourceTester(pulsarCluster, converterClassName); + sourceTester.getSourceConfig().put("json-with-envelope", jsonWithEnvelope); // setup debezium mysql server DebeziumMySQLContainer mySQLContainer = new DebeziumMySQLContainer(pulsarCluster.getClusterName()); @@ -2330,26 +2338,35 @@ private void testDebeziumMySqlConnect() Failsafe.with(statusRetryPolicy).run(() -> waitForProcessingSourceMessages(tenant, namespace, sourceName, numMessages)); + @Cleanup + Consumer consumer = client.newConsumer(getSchema(jsonWithEnvelope)) + .topic(consumeTopicName) + .subscriptionName("debezium-source-tester") + .subscriptionType(SubscriptionType.Exclusive) + .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest) + .subscribe(); + log.info("[debezium mysql test] create consumer finish. converterName: {}", converterClassName); + // validate the source result - sourceTester.validateSourceResult(consumer, 9, null); + sourceTester.validateSourceResult(consumer, 9, null, converterClassName); // prepare insert event sourceTester.prepareInsertEvent(); // validate the source insert event - sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT); + sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT, converterClassName); // prepare update event sourceTester.prepareUpdateEvent(); // validate the source update event - sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE, converterClassName); // prepare delete event sourceTester.prepareDeleteEvent(); // validate the source delete event - sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE, converterClassName); // delete the source deleteSource(tenant, namespace, sourceName); @@ -2358,14 +2375,14 @@ private void testDebeziumMySqlConnect() getSourceInfoNotFound(tenant, namespace, sourceName); } - private void testDebeziumPostgreSqlConnect() throws Exception { + private void testDebeziumPostgreSqlConnect(String converterClassName, boolean jsonWithEnvelope) throws Exception { final String tenant = TopicName.PUBLIC_TENANT; final String namespace = TopicName.DEFAULT_NAMESPACE; final String outputTopicName = "debe-output-topic-name"; - final String consumeTopicName = "public/default/dbserver1.inventory.products"; - final String sourceName = "test-source-connector-" - + functionRuntimeType + "-name-" + randomName(8); + final String consumeTopicName = "debezium/postgresql/dbserver1.inventory.products"; + final String sourceName = "test-source-debezium-postgersql-" + functionRuntimeType + "-" + randomName(8); + // This is the binlog count that contained in postgresql container. final int numMessages = 26; @@ -2382,21 +2399,13 @@ private void testDebeziumPostgreSqlConnect() throws Exception { @Cleanup PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build(); - try { - // If topic already exists, we should delete it so as not to affect the following tests. - admin.topics().getStats(consumeTopicName); - admin.topics().delete(consumeTopicName); - admin.schemas().deleteSchema(consumeTopicName); - } catch (PulsarAdminException e) { - // Expected results, ignoring the exception - log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}", - consumeTopicName, e.getMessage()); - } + initNamespace(admin); + admin.topics().createNonPartitionedTopic(consumeTopicName); admin.topics().createNonPartitionedTopic(outputTopicName); @Cleanup - Consumer> consumer = client.newConsumer(KeyValueSchema.kvBytes()) + Consumer consumer = client.newConsumer(getSchema(jsonWithEnvelope)) .topic(consumeTopicName) .subscriptionName("debezium-source-tester") .subscriptionType(SubscriptionType.Exclusive) @@ -2404,6 +2413,7 @@ private void testDebeziumPostgreSqlConnect() throws Exception { @Cleanup DebeziumPostgreSqlSourceTester sourceTester = new DebeziumPostgreSqlSourceTester(pulsarCluster); + sourceTester.getSourceConfig().put("json-with-envelope", jsonWithEnvelope); // setup debezium postgresql server DebeziumPostgreSqlContainer postgreSqlContainer = new DebeziumPostgreSqlContainer(pulsarCluster.getClusterName()); @@ -2426,25 +2436,25 @@ private void testDebeziumPostgreSqlConnect() throws Exception { waitForProcessingSourceMessages(tenant, namespace, sourceName, numMessages)); // validate the source result - sourceTester.validateSourceResult(consumer, 9, null); + sourceTester.validateSourceResult(consumer, 9, null, converterClassName); // prepare insert event sourceTester.prepareInsertEvent(); // validate the source insert event - sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT); + sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT, converterClassName); // prepare update event sourceTester.prepareUpdateEvent(); // validate the source update event - sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE, converterClassName); // prepare delete event sourceTester.prepareDeleteEvent(); // validate the source delete event - sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE, converterClassName); // delete the source deleteSource(tenant, namespace, sourceName); @@ -2453,12 +2463,12 @@ private void testDebeziumPostgreSqlConnect() throws Exception { getSourceInfoNotFound(tenant, namespace, sourceName); } - private void testDebeziumMongoDbConnect() throws Exception { + private void testDebeziumMongoDbConnect(String converterClassName, boolean jsonWithEnvelope) throws Exception { final String tenant = TopicName.PUBLIC_TENANT; final String namespace = TopicName.DEFAULT_NAMESPACE; final String outputTopicName = "debe-output-topic-name"; - final String consumeTopicName = "public/default/dbserver1.inventory.products"; + final String consumeTopicName = "debezium/mongodb/dbserver1.inventory.products"; final String sourceName = "test-source-connector-" + functionRuntimeType + "-name-" + randomName(8); @@ -2477,21 +2487,13 @@ private void testDebeziumMongoDbConnect() throws Exception { @Cleanup PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(pulsarCluster.getHttpServiceUrl()).build(); - try { - // If topic already exists, we should delete it so as not to affect the following tests. - admin.topics().getStats(consumeTopicName); - admin.topics().delete(consumeTopicName); - admin.schemas().deleteSchema(consumeTopicName); - } catch (PulsarAdminException e) { - // Expected results, ignoring the exception - log.info("Topic: {} does not exist, we can continue the following tests. Exceptions message: {}", - consumeTopicName, e.getMessage()); - } + initNamespace(admin); + admin.topics().createNonPartitionedTopic(consumeTopicName); admin.topics().createNonPartitionedTopic(outputTopicName); @Cleanup - Consumer> consumer = client.newConsumer(KeyValueSchema.kvBytes()) + Consumer consumer = client.newConsumer(getSchema(jsonWithEnvelope)) .topic(consumeTopicName) .subscriptionName("debezium-source-tester") .subscriptionType(SubscriptionType.Exclusive) @@ -2499,6 +2501,7 @@ private void testDebeziumMongoDbConnect() throws Exception { @Cleanup DebeziumMongoDbSourceTester sourceTester = new DebeziumMongoDbSourceTester(pulsarCluster); + sourceTester.getSourceConfig().put("json-with-envelope", jsonWithEnvelope); // setup debezium mongodb server DebeziumMongoDbContainer mongoDbContainer = new DebeziumMongoDbContainer(pulsarCluster.getClusterName()); @@ -2520,25 +2523,25 @@ private void testDebeziumMongoDbConnect() throws Exception { waitForProcessingSourceMessages(tenant, namespace, sourceName, numMessages)); // validate the source result - sourceTester.validateSourceResult(consumer, 9, null); + sourceTester.validateSourceResult(consumer, 9, null, converterClassName); // prepare insert event sourceTester.prepareInsertEvent(); // validate the source insert event - sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT); + sourceTester.validateSourceResult(consumer, 1, SourceTester.INSERT, converterClassName); // prepare update event sourceTester.prepareUpdateEvent(); // validate the source update event - sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.UPDATE, converterClassName); // prepare delete event sourceTester.prepareDeleteEvent(); // validate the source delete event - sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE); + sourceTester.validateSourceResult(consumer, 1, SourceTester.DELETE, converterClassName); // delete the source deleteSource(tenant, namespace, sourceName); @@ -2547,4 +2550,27 @@ private void testDebeziumMongoDbConnect() throws Exception { getSourceInfoNotFound(tenant, namespace, sourceName); } + private void initNamespace(PulsarAdmin admin) { + log.info("[initNamespace] start."); + try { + admin.tenants().createTenant("debezium", new TenantInfo(Sets.newHashSet(), + Sets.newHashSet(pulsarCluster.getClusterName()))); + admin.namespaces().createNamespace("debezium/mysql-json"); + admin.namespaces().createNamespace("debezium/mysql-avro"); + admin.namespaces().createNamespace("debezium/mongodb"); + admin.namespaces().createNamespace("debezium/postgresql"); + } catch (Exception e) { + log.info("[initNamespace] msg: {}", e.getMessage()); + } + log.info("[initNamespace] finish."); + } + + private Schema getSchema(boolean jsonWithEnvelope) { + if (jsonWithEnvelope) { + return KeyValueSchema.kvBytes(); + } else { + return KeyValueSchema.of(Schema.AUTO_CONSUME(), Schema.AUTO_CONSUME(), KeyValueEncodingType.SEPARATED); + } + } + } diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMongoDbSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMongoDbSourceTester.java index 23b5db43134e4..6fa35ef918376 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMongoDbSourceTester.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMongoDbSourceTester.java @@ -50,6 +50,7 @@ public DebeziumMongoDbSourceTester(PulsarCluster cluster) { sourceConfig.put("mongodb.task.id","1"); sourceConfig.put("database.whitelist", "inventory"); sourceConfig.put("pulsar.service.url", pulsarServiceUrl); + sourceConfig.put("topic.namespace", "debezium/mongodb"); } @Override diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java index 3287e2b0750c2..4b665065e0e01 100644 --- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java +++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java @@ -48,7 +48,7 @@ public class DebeziumMySqlSourceTester extends SourceTester { protected final String sourceType; protected final Map sourceConfig; + public final static Set DEBEZIUM_FIELD_SET = new HashSet() {{ + add("before"); + add("after"); + add("source"); + add("op"); + add("ts_ms"); + }}; + protected SourceTester(String sourceType) { this.sourceType = sourceType; this.sourceConfig = Maps.newHashMap(); @@ -71,7 +83,16 @@ public Map sourceConfig() { public abstract Map produceSourceMessages(int numMessages) throws Exception; - public void validateSourceResult(Consumer> consumer, int number, String eventType) throws Exception { + public void validateSourceResult(Consumer consumer, int number, + String eventType, String converterClassName) throws Exception { + if (converterClassName.endsWith("AvroConverter")) { + validateSourceResultAvro(consumer, number, eventType); + } else { + validateSourceResultJson(consumer, number, eventType); + } + } + + public void validateSourceResultJson(Consumer> consumer, int number, String eventType) throws Exception { int recordsNumber = 0; Message> msg = consumer.receive(2, TimeUnit.SECONDS); while(msg != null) { @@ -82,7 +103,36 @@ public void validateSourceResult(Consumer> consumer, in Assert.assertTrue(key.contains(this.keyContains())); Assert.assertTrue(value.contains(this.valueContains())); if (eventType != null) { - Assert.assertTrue(value.contains(this.eventContains(eventType))); + Assert.assertTrue(value.contains(this.eventContains(eventType, true))); + } + consumer.acknowledge(msg); + msg = consumer.receive(1, TimeUnit.SECONDS); + } + + Assert.assertEquals(recordsNumber, number); + log.info("Stop {} server container. topic: {} has {} records.", getSourceType(), consumer.getTopic(), recordsNumber); + } + + public void validateSourceResultAvro(Consumer> consumer, + int number, String eventType) throws Exception { + int recordsNumber = 0; + Message> msg = consumer.receive(2, TimeUnit.SECONDS); + while(msg != null) { + recordsNumber ++; + GenericRecord keyRecord = msg.getValue().getKey(); + Assert.assertNotNull(keyRecord.getFields()); + Assert.assertTrue(keyRecord.getFields().size() > 0); + + GenericRecord valueRecord = msg.getValue().getValue(); + Assert.assertNotNull(valueRecord.getFields()); + Assert.assertTrue(valueRecord.getFields().size() > 0); + for (Field field : valueRecord.getFields()) { + Assert.assertTrue(DEBEZIUM_FIELD_SET.contains(field.getName())); + } + + if (eventType != null) { + String op = valueRecord.getField("op").toString(); + Assert.assertEquals(this.eventContains(eventType, false), op); } consumer.acknowledge(msg); msg = consumer.receive(1, TimeUnit.SECONDS); @@ -91,20 +141,22 @@ public void validateSourceResult(Consumer> consumer, in Assert.assertEquals(recordsNumber, number); log.info("Stop {} server container. topic: {} has {} records.", getSourceType(), consumer.getTopic(), recordsNumber); } + public String keyContains(){ return "dbserver1.inventory.products.Key"; } + public String valueContains(){ return "dbserver1.inventory.products.Value"; } - public String eventContains(String eventType) { + public String eventContains(String eventType, boolean isJson) { if (eventType.equals(INSERT)) { - return "\"op\":\"c\""; + return isJson ? "\"op\":\"c\"" : "c"; } else if (eventType.equals(UPDATE)) { - return "\"op\":\"u\""; + return isJson ? "\"op\":\"u\"" : "u"; } else { - return "\"op\":\"d\""; + return isJson ? "\"op\":\"d\"" : "d"; } } }