[Pulsar IO][Issue 5633]Support avro schema for debezium connector #6034
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
You are right, the current integration test has been covered, and this feature supports JSONSchema and AVROSchema. @sijie |
sijie
requested changes
Feb 6, 2020
| @@ -188,15 +198,48 @@ public void close() { | |||
| @Getter | |||
| Optional<String> destinationTopic; | |||
|
|
|||
| private final AvroData avroData; | |||
|
|
|||
| private org.apache.pulsar.kafka.shade.avro.Schema keyAvroSchema; | |||
There was a problem hiding this comment.
I don't think this needs to be a class variable. what method is using this variable except the constructor?
|
|
||
| private org.apache.pulsar.kafka.shade.avro.Schema keyAvroSchema; | ||
|
|
||
| private org.apache.pulsar.kafka.shade.avro.Schema valueAvroSchema; |
|
|
||
| private final KafkaSchema valueSchema; | ||
|
|
||
| private byte[] keyBytes; |
There was a problem hiding this comment.
why do we need to make this variable a class variable? it is only used in the constructor.
|
|
||
| private byte[] keyBytes; | ||
|
|
||
| private byte[] valueBytes; |
| String keyName = this.topicName.get() + "-key"; | ||
| String valueName = this.topicName.get() + "-value"; | ||
|
|
||
| if (readerCache.getIfPresent(keyName) == null |
There was a problem hiding this comment.
actually I think the cache here is incorrect. you need a cache from kafka connect schema to pulsar schema.
Cache<org.apache.kafka.connect.data.Schema, KafkaSchema>
|
I have tried to build and run it. I have made sure to delete all the existing schemas before running so the schema mutation error is not stem from leftover schema from previous runs. And if I use AvroConverter (org.apache.pulsar.kafka.shade.io.confluent.connect.avro.AvroConverter) |
|
@tuteng What is the status for this PR? |
|
@dennisylyung Hi, could you provide some config info about the error logs, such as the pulsar source config and the reproduce steps, thanks. This is my test steps to follow the demo http://pulsar.apache.org/docs/en/io-debezium-source/#example-of-mysql, but I didn't reproduce your problem. |
|
@gaoran10 is helping on this issue. |
This is the config I was using. |
gaoran10
force-pushed
the
feature/debezium-support-avro-schema
branch
from
f3f1974
to
a94785f
Compare
|
I have made some changes, use the JsonConverter as before, and the AvroConverter is used differently from the JsonConverter, the user has two ways to use it.
@tuteng @sijie @jiazhai @codelipenghui Please have a review. Thanks. |
codelipenghui
requested changes
Apr 18, 2020
There was a problem hiding this comment.
I don't see how users pass offload to the schema info. Currently, users use SchemaDefination to create schemas, so it's better to add a method in the SchemaDefinationBuilder.
And, we also need a doc to tell users how to consume messages from the topic that the debezium writes. If we have debezium connector related docs, it's better to add them directly. Otherwise, you can create a issue or a new doc for debezium connector.
| @@ -47,6 +47,10 @@ | |||
| private final List<Field> fields; | |||
| private final Schema schema; | |||
| private final byte[] schemaVersion; | |||
|
|
|||
| private int offset; | |||
| public final static String OFFSET_PROP = "AVRO_READ_OFFSET"; | |||
There was a problem hiding this comment.
There are two properties name definition in the SchemaDefinitionBuilderImpl. I think you can move all of them to SchemaDefinition and keep the with prefix "__".
codelipenghui
added
the
doc-required
|
/pulsarbot run-failure-checks |
gaoran10
force-pushed
the
feature/debezium-support-avro-schema
branch
from
1e71e0a
to
4e077b5
Compare
|
/pulsarbot run-failure-checks |
2 similar comments
|
/pulsarbot run-failure-checks |
|
/pulsarbot run-failure-checks |
jiazhai
approved these changes
Apr 29, 2020
sijie
approved these changes
Apr 29, 2020
codelipenghui
approved these changes
Apr 30, 2020
Huanli-Meng
pushed a commit
to Huanli-Meng/pulsar
that referenced
this pull request
May 27, 2020
…ache#6034) Fixes apache#5633 ### Motivation Currently, some users want to support Avro schema in debezium, so this pr supports this feature. For Kafka's Avro schema, it depends on the Avro 1.8 version, but Avro version has just been upgraded to 1.9 in pulsar, so shade is needed to avoid finding `addProp` function ### Modifications * Add a package `kafka-connect-avro-converter-shaded` * Add class KafkaSchema to converter Kafka's Avro schema to pulsar's schema ### Verifying this change Unit test and integration tests
Anonymitaet
removed
the
doc-required
|
Doc has been added: https://pulsar.apache.org/docs/en/next/io-debezium-source/#converter-options |
huangdx0726
pushed a commit
to huangdx0726/pulsar
that referenced
this pull request
Aug 24, 2020
…ache#6034) Fixes apache#5633 ### Motivation Currently, some users want to support Avro schema in debezium, so this pr supports this feature. For Kafka's Avro schema, it depends on the Avro 1.8 version, but Avro version has just been upgraded to 1.9 in pulsar, so shade is needed to avoid finding `addProp` function ### Modifications * Add a package `kafka-connect-avro-converter-shaded` * Add class KafkaSchema to converter Kafka's Avro schema to pulsar's schema ### Verifying this change Unit test and integration tests
14 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Fixes #5633
Motivation
Currently, some users want to support Avro schema in debezium, so this pr supports this feature.
For Kafka's Avro schema, it depends on the Avro 1.8 version, but Avro version has just been upgraded to 1.9 in pulsar, so shade is needed to avoid finding
addPropfunctionModifications
kafka-connect-avro-converter-shadedVerifying this change
Unit test and integration tests