[Feature][seatunnel-connectors-v2][connector-kafka] Kafka supports custom schema #2371 #2720
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
TyrantLucifer
requested changes
Sep 12, 2022
| @@ -0,0 +1,51 @@ | |||
| /* | |||
There was a problem hiding this comment.
seatunnel-examples module is only used by developers. So test cases should not be submitted
| public class KafkaSource implements SeaTunnelSource<SeaTunnelRow, KafkaSourceSplit, KafkaSourceState> { | ||
|
|
||
| private static final String DEFAULT_CONSUMER_GROUP = "SeaTunnel-Consumer-Group"; | ||
| public class KafkaSource<T> implements SeaTunnelSource<T, KafkaSourceSplit, KafkaSourceState> { |
There was a problem hiding this comment.
Why change the generic type?
There was a problem hiding this comment.
Refer to the PULSAR implementation
| } | ||
|
|
||
| @Override | ||
| public SeaTunnelRowType getProducedType() { | ||
| return this.typeInfo; | ||
| public SeaTunnelDataType<T> getProducedType() { |
| } | ||
|
|
||
| @Override | ||
| public SourceReader<SeaTunnelRow, KafkaSourceSplit> createReader(SourceReader.Context readerContext) throws Exception { | ||
| return new KafkaSourceReader(this.metadata, this.typeInfo, readerContext); | ||
| public SourceReader<T, KafkaSourceSplit> createReader(SourceReader.Context readerContext) throws Exception { |
| } else { | ||
| rowType = SeaTunnelSchema.buildSimpleTextSchema(); | ||
| } | ||
| deserializationSchema = (DeserializationSchema<T>) new JsonDeserializationSchema(false, false, rowType); |
There was a problem hiding this comment.
T should replaced by SeaTunnelRow
| @@ -46,7 +47,7 @@ | |||
| import java.util.Set; | |||
| import java.util.stream.Collectors; | |||
|
|
|||
| public class KafkaSourceReader implements SourceReader<SeaTunnelRow, KafkaSourceSplit> { | |||
| public class KafkaSourceReader<T> implements SourceReader<T, KafkaSourceSplit> { | |||
| // TODO support user custom type | ||
| private SeaTunnelRowType typeInfo; | ||
|
|
||
| private final DeserializationSchema<T> deserializationSchema; |
| @@ -87,7 +89,7 @@ public void close() throws IOException { | |||
| } | |||
|
|
|||
| @Override | |||
| public void pollNext(Collector<SeaTunnelRow> output) throws Exception { | |||
| public void pollNext(Collector<T> output) throws Exception { | |||
| }else{ | ||
| String v = stringDeserializer.deserialize(partition.topic(), record.value()); | ||
| String t = partition.topic(); | ||
| output.collect((T) new SeaTunnelRow(new Object[]{t, v})); |
There was a problem hiding this comment.
If user does not assign schema, it will generate a simple schema as the following shown:
content
data
So the row only have one field named content, but you generate a row with two fields.
|
BTW, you should add spark-e2e test cases and update doc of kafka source conenctor too. |
ashulin
reviewed
Sep 12, 2022
hailin0
reviewed
Sep 13, 2022
Comment on lines
+33
to
+34
| name = "string" | ||
| age = "int" |
This was referenced Sep 18, 2022
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.
Purpose of this pull request
[Feature][seatunnel-connectors-v2][connector-kafka] Kafka supports custom schema #2371
Check list
New License Guide
此拉取请求的目的
Kafka 支持自定义Schema #2371
检查列表
添加e2e测试
添加example测试