Add KeyValueSchema to satisfy org.apache.kafka.connect.connector.ConnectRecord #2885
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
sijie
added
type/feature
sijie
requested changes
Oct 29, 2018
| /** | ||
| * Key Value Schema whose underneath schemas are AvroSchema. | ||
| */ | ||
| static Schema<?> KeyValue(Class key, Class value) { |
There was a problem hiding this comment.
actually I think it might be better to do following:
- be able to construct keyvalue schema with
SchemaType - make the default as JSON, since we are using JSON as default schema for other places.
static <K, V> Schema<KeyValue<K, V>> KeyValue(Class<K> key, Class<V> value, SchemaType type) {
checkArgument(SchemaType.JSON == type || SchemaType.AVRO == type);
...
}
static <K, V> Schema<KeyValue<K, V>> KeyValue(Class<K> key, Class<V> value) {
return KeyValue(key, value, SchemaType.JSON);
...
}
| return new KeyValueSchema<>(AvroSchema.of(key), AvroSchema.of(value)); | ||
| } | ||
|
|
||
| static Schema<?> KeyValue(Schema key, Schema value) { |
There was a problem hiding this comment.
static <K, V> Schema<KeyValue<K, V>> KeyValue(Schema<K> key, Schema<V> value);
| @Slf4j | ||
| public class KeyValueSchema<K, V> implements Schema<KeyValue<K, V>> { | ||
| @Getter | ||
| private Schema<K> keySchema; |
| @Getter | ||
| private Schema<K> keySchema; | ||
| @Getter | ||
| private Schema<V> valueSchema; |
| return value; | ||
| } | ||
|
|
||
| public void setKey(K key) { |
There was a problem hiding this comment.
do we need to setKey or setValue? If not, I would start with making key and value immutable.
|
rerun java8 tests |
sijie
approved these changes
Oct 30, 2018
|
|
||
| // schemaInfo combined by KeySchemaInfo and ValueSchemaInfo: | ||
| // [keyInfo.length][keyInfo][valueInfo.length][ValueInfo] | ||
| private SchemaInfo schemaInfo; |
|
run integration tests |
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.
Kafka org.apache.kafka.connect.connector.ConnectRecord has both key and value schema.
Currently we need a KeyValueSchema to achieve it and support debezium.
There will be another PR to do the convert between KeyValueSchema and org.apache.kafka.connect.data.Schema.