From 404418649fb879fd32063de5057bd6411fec1bee Mon Sep 17 00:00:00 2001
From: Etienne Chauchot <echauchot@apache.org>
Date: Wed, 4 Apr 2018 17:03:32 +0200
Subject: [PATCH] [BEAM-2852] Fix the code that uses Long key whereas there is
 no key in the PCollection

---
 .../beam/sdk/nexmark/NexmarkLauncher.java     | 25 +++++++++----------
 1 file changed, 12 insertions(+), 13 deletions(-)

diff --git a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
index cb8820f62a27..a7218aae635d 100644
--- a/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
+++ b/sdks/java/nexmark/src/main/java/org/apache/beam/sdk/nexmark/NexmarkLauncher.java
@@ -88,20 +88,17 @@
 import org.apache.beam.sdk.nexmark.queries.sql.SqlQuery7;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.util.CoderUtils;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionTuple;
-import org.apache.beam.sdk.values.PDone;
 import org.apache.beam.sdk.values.TimestampedValue;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.TupleTagList;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.kafka.common.serialization.LongDeserializer;
-import org.apache.kafka.common.serialization.LongSerializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.joda.time.Duration;
 import org.slf4j.LoggerFactory;
@@ -792,12 +789,13 @@ public void processElement(ProcessContext c) {
    * Send {@code events} to Kafka.
    */
   private void sinkEventsToKafka(PCollection<Event> events) {
-    PTransform<PCollection<byte[]>, PDone> io = KafkaIO.<Long, byte[]>write()
-            .withBootstrapServers(options.getBootstrapServers())
-            .withTopic(options.getKafkaSinkTopic())
-            .withKeySerializer(LongSerializer.class)
-            .withValueSerializer(ByteArraySerializer.class).values();
-    events.apply("Event to bytes", ParDo.of(EVENT_TO_BYTEARRAY)).apply(io);
+    PCollection<byte[]> event_to_bytes = events.apply("Event to bytes", ParDo.of(EVENT_TO_BYTEARRAY));
+    event_to_bytes.apply(KafkaIO.<Void, byte[]>write()
+                    .withBootstrapServers(options.getBootstrapServers())
+                    .withTopic(options.getKafkaSinkTopic())
+                    .withValueSerializer(ByteArraySerializer.class)
+                    .values());
+
   }
 
 
@@ -892,12 +890,13 @@ private void sinkResultsToKafka(PCollection<String> formattedResults) {
     checkArgument(!Strings.isNullOrEmpty(options.getBootstrapServers()),
             "Missing --bootstrapServers");
 
-    PTransform<PCollection<String>, PDone> io = KafkaIO.<Long, String>write()
+    formattedResults.apply(
+        queryName + ".WriteKafkaResults",
+        KafkaIO.<Void, String>write()
             .withBootstrapServers(options.getBootstrapServers())
             .withTopic(options.getKafkaSinkTopic())
-            .withKeySerializer(LongSerializer.class)
-            .withValueSerializer(StringSerializer.class).values();
-    formattedResults.apply(queryName + ".WriteKafkaResults", io);
+            .withValueSerializer(StringSerializer.class)
+            .values());
   }
 
   /**