[beam] branch master updated: [BEAM-3754]: Fix readBytes() initialization (#4792)

[mi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mingmxu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new befd977  [BEAM-3754]: Fix readBytes() initialization (#4792)
befd977 is described below

commit befd97741423f174998bbcc123b2ada21590a1fb
Author: Raghu Angadi <ra...@apache.org>
AuthorDate: Sat Mar 3 21:47:09 2018 -0800

    [BEAM-3754]: Fix readBytes() initialization (#4792)
---
 .../java/org/apache/beam/sdk/io/kafka/KafkaIO.java     | 18 +++++-------------
 .../java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java |  6 ++----
 2 files changed, 7 insertions(+), 17 deletions(-)

diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index cc286e9..eb29229 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -221,21 +221,13 @@ import org.slf4j.LoggerFactory;
 public class KafkaIO {
 
   /**
-   * Creates an uninitialized {@link Read} {@link PTransform}. Before use, basic Kafka
-   * configuration should set with {@link Read#withBootstrapServers(String)} and
-   * {@link Read#withTopics(List)}. Other optional settings include key and value
-   * {@link Deserializer}s, custom timestamp and watermark functions.
+   * A specific instance of uninitialized {@link #read()} where key and values are bytes.
+   * See #read().
    */
   public static Read<byte[], byte[]> readBytes() {
-    return new AutoValue_KafkaIO_Read.Builder<byte[], byte[]>()
-        .setTopics(new ArrayList<>())
-        .setTopicPartitions(new ArrayList<>())
-        .setKeyDeserializer(ByteArrayDeserializer.class)
-        .setValueDeserializer(ByteArrayDeserializer.class)
-        .setConsumerFactoryFn(Read.KAFKA_CONSUMER_FACTORY_FN)
-        .setConsumerConfig(Read.DEFAULT_CONSUMER_PROPERTIES)
-        .setMaxNumRecords(Long.MAX_VALUE)
-        .build();
+    return KafkaIO.<byte[], byte[]>read()
+      .withKeyDeserializer(ByteArrayDeserializer.class)
+      .withValueDeserializer(ByteArrayDeserializer.class);
   }
 
   /**
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
index d3f8f7d..3718c41 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
@@ -1044,14 +1044,12 @@ public class KafkaIOTest {
 
   @Test
   public void testSourceWithExplicitPartitionsDisplayData() {
-    KafkaIO.Read<byte[], Long> read = KafkaIO.<byte[], Long>read()
+    KafkaIO.Read<byte[], byte[]> read = KafkaIO.readBytes()
         .withBootstrapServers("myServer1:9092,myServer2:9092")
         .withTopicPartitions(ImmutableList.of(new TopicPartition("test", 5),
             new TopicPartition("test", 6)))
         .withConsumerFactoryFn(new ConsumerFactoryFn(
-            Lists.newArrayList("test"), 10, 10, OffsetResetStrategy.EARLIEST)) // 10 partitions
-        .withKeyDeserializer(ByteArrayDeserializer.class)
-        .withValueDeserializer(LongDeserializer.class);
+            Lists.newArrayList("test"), 10, 10, OffsetResetStrategy.EARLIEST)); // 10 partitions
 
     DisplayData displayData = DisplayData.from(read);
 

-- 
To stop receiving notification emails like this one, please contact
mingmxu@apache.org.