make BeamKafkaTable.createKafkaRead to be protected, and we can overr…

…ide it appending options to KafkaIO.Read (#29051)
apache · Oct 19, 2023 · 67a39a5 · 67a39a5
1 parent 39678b6
commit 67a39a5
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/.../src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java b/.../src/main/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/BeamKafkaTable.java
@@ -110,7 +110,7 @@ public PCollection<Row> buildIOReader(PBegin begin) {
         .setRowSchema(getSchema());
   }
 
-  KafkaIO.Read<byte[], byte[]> createKafkaRead() {
+  protected KafkaIO.Read<byte[], byte[]> createKafkaRead() {
     KafkaIO.Read<byte[], byte[]> kafkaRead;
     if (topics != null) {
       kafkaRead =

diff --git a/.../src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTestTable.java b/.../src/test/java/org/apache/beam/sdk/extensions/sql/meta/provider/kafka/KafkaTestTable.java
@@ -61,7 +61,7 @@ public KafkaTestTable(Schema beamSchema, List<String> topics, int partitionsPerT
   }
 
   @Override
-  KafkaIO.Read<byte[], byte[]> createKafkaRead() {
+  protected KafkaIO.Read<byte[], byte[]> createKafkaRead() {
     return super.createKafkaRead().withConsumerFactoryFn(this::mkMockConsumer);
   }