to #103, Support PB format serializer in Kafka Source #185

qidian99 · 2022-11-17T18:30:26Z

Signed-off-by:

to #103, Support PB format serializer in Kafka Source

garyli1019 · 2022-11-18T01:43:52Z

hi @qidian99 , thanks for your contribution, looks like there is a conflict. Would you rebase the latest master?

qidian99 · 2022-11-18T04:14:42Z

hi @qidian99 , thanks for your contribution, looks like there is a conflict. Would you rebase the latest master?

Done rebasing. @garyli1019

garyli1019

Thanks for your contribution! left some comments

garyli1019 · 2022-11-18T06:31:28Z

...-kafka/src/test/java/com/bytedance/bitsail/connector/legacy/kafka/common/KafkaPbMessage.java

@@ -0,0 +1,970 @@
+


can we add the license header?

This class is generated by protoc, which isn't dynamic. I removed it and made the test case directly read proto descriptor for job configuration.

I think it's ok to put this class in our codebase, because if we generate this class from the test case, this class will be recognized by git and added to the git commit for those who run the test case. It's tricky to add this to .gitignore as well. Add the license header on an auto generated file is fine I believe.

nvm, my concern didn't happen at all.

garyli1019 · 2022-11-18T06:38:49Z

...m/bytedance/bitsail/connector/legacy/kafka/deserialization/DeserializationSchemaFactory.java

@@ -64,6 +65,14 @@ public static KafkaDeserializationSchema<Row> getDeserializationSchema(BitSailCo
              .build());
    }

+    if (StringUtils.equalsIgnoreCase(PB_DESERIALIZATION_SCHEMA_KEY, formatType)) {
+      try {
+        return new PbKafkaDeserializationSchema(configuration);


can we use the CountKafkaDeserializationSchemaWrapper like json, this wrapper is useful to run our regression test for streaming mode.

Done, PbKafkaDeserializationSchema is renamed to PbDeserializationSchema and extends Flink's deserialization schema

garyli1019 · 2022-11-18T06:41:42Z

...m/bytedance/bitsail/connector/legacy/kafka/deserialization/PbKafkaDeserializationSchema.java

+import java.util.List;
+
+@Internal
+public class PbKafkaDeserializationSchema implements KafkaDeserializationSchema<Row> {


with the wrapper, this class could be more general which not only work for kafka. Can we move this to bitsail-components/bitsail-component-formats/bitsail-component-format-pb

Tried generalizing PbDeserializationSchema. Already created empty module named bitsail-component-format-pb. However, there are some problems:

the wrapper takes in org.apache.flink.api.common.serialization.DeserializationSchema while deserialization schemas in bitsail-component-formats implements com.bytedance.bitsail.base.format.DeserializationSchema<I, O>

v1 connectors are not using spi to get deserialization schemas, e.g., RocketMQDeserializationSchema

IMHO, It is better to create another issue for bitsail-component-formats implementations

Got it, I agree with you here! Please feel free to create an issue if you're interested with the follow up.

garyli1019 · 2022-11-18T06:43:12Z

...m/bytedance/bitsail/connector/legacy/kafka/deserialization/PbKafkaDeserializationSchema.java

+  }
+
+  @Override
+  public Row deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {


can we can UT for this, I believe all the supported types should be covered

garyli1019 · 2022-11-18T06:45:26Z

...fka/src/test/java/com/bytedance/bitsail/connector/legacy/kafka/source/KafkaSourceITCase.java

+    }, 0, 1, TimeUnit.SECONDS);
+  }
+
+  @SneakyThrows


not sure why we need this SneakyThrows here, would you elaborate？

Removed. Thanks for pointing out!

garyli1019 · 2022-11-18T06:49:05Z

...-connectors/bitsail-connectors-legacy/bitsail-connector-kafka/src/test/resources/kafka.proto

@@ -0,0 +1,7 @@
+syntax = "proto3";
+
+message ProtoTest {


I'd suggest we include all supported types here, including some complex type.

garyli1019 · 2022-11-18T06:51:03Z

...l-connectors-legacy/bitsail-connector-kafka/src/test/resources/kafka_to_print_pb_format.json

+        "update-mode": "append"
+      },
+      "proto": {
+        "descriptor": "",


This json sample should be runnable. It's ok to add the pb info from the integration test you write.

Added base64 encoded descriptor and class name

garyli1019 · 2022-11-21T02:38:56Z

...fka/src/test/java/com/bytedance/bitsail/connector/legacy/kafka/source/KafkaSourceITCase.java

      }
    }, 0, 1, TimeUnit.SECONDS);
  }

  @Test
-  public void testKafkaSource() throws Exception {
+  public void testKafkaSourceJsonFormat() throws Exception {


I ran this ITCase and one test was failed. looks like the kafka producer didn't stop when we switch to the second test.

Added clean up logic on the producer service to stop creating records. All test cases pass now

BlockLiu · 2022-12-26T12:25:34Z

👏 Great work!
I see there are dependency conflicts leading to compiling failure, would you update the PR and resolve this conflict?

qidian99 mentioned this pull request Nov 17, 2022

[BitSail][Connector] Support PB format serializer in Kafka Source #103

Open

garyli1019 self-assigned this Nov 18, 2022

qidian99 force-pushed the issue/#103_support_pb_format_in_kafka branch 2 times, most recently from 2cb8936 to 40983a3 Compare November 18, 2022 04:12

qidian99 force-pushed the issue/#103_support_pb_format_in_kafka branch from 40983a3 to bc60582 Compare November 18, 2022 06:45

garyli1019 reviewed Nov 18, 2022

View reviewed changes

garyli1019 reviewed Nov 21, 2022

View reviewed changes

qidian99 force-pushed the issue/#103_support_pb_format_in_kafka branch from 87e5e04 to a46ffa4 Compare November 22, 2022 04:35

to bytedance#103, Support PB format serializer in Kafka Source

00fca9e

qidian99 force-pushed the issue/#103_support_pb_format_in_kafka branch from a46ffa4 to 00fca9e Compare November 22, 2022 04:40

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

to #103, Support PB format serializer in Kafka Source #185

to #103, Support PB format serializer in Kafka Source #185

qidian99 commented Nov 17, 2022

garyli1019 commented Nov 18, 2022

qidian99 commented Nov 18, 2022 •

edited

Loading

garyli1019 left a comment

garyli1019 Nov 18, 2022

qidian99 Nov 19, 2022

garyli1019 Nov 21, 2022

garyli1019 Nov 21, 2022

garyli1019 Nov 18, 2022

qidian99 Nov 19, 2022

garyli1019 Nov 18, 2022

qidian99 Nov 19, 2022 •

edited

Loading

garyli1019 Nov 21, 2022

garyli1019 Nov 18, 2022

garyli1019 Nov 18, 2022

qidian99 Nov 19, 2022

garyli1019 Nov 18, 2022

garyli1019 Nov 18, 2022

qidian99 Nov 19, 2022

garyli1019 Nov 21, 2022

qidian99 Nov 22, 2022

BlockLiu commented Dec 26, 2022

to #103, Support PB format serializer in Kafka Source #185

Are you sure you want to change the base?

to #103, Support PB format serializer in Kafka Source #185

Conversation

qidian99 commented Nov 17, 2022

garyli1019 commented Nov 18, 2022

qidian99 commented Nov 18, 2022 • edited Loading

garyli1019 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qidian99 Nov 19, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BlockLiu commented Dec 26, 2022

qidian99 commented Nov 18, 2022 •

edited

Loading

qidian99 Nov 19, 2022 •

edited

Loading