[Streaming][Kafka][SPARK-8390] doc changes per TD, test to make sure approach shown in docs actually compiles + runs

Author: koeninger

docs/streaming-kafka-integration.md

@@ -82,7 +82,7 @@ This approach has the following advantages over the receiver-based approach (i.e
 
 - *Efficiency:* Achieving zero-data loss in the first approach required the data to be stored in a Write Ahead Log, which further replicated the data. This is actually inefficient as the data effectively gets replicated twice - once by Kafka, and a second time by the Write Ahead Log. This second approach eliminates the problem as there is no receiver, and hence no need for Write Ahead Logs. As long as you have sufficient Kafka retention, messages can be recovered from Kafka.
 
-- *Exactly-once semantics:* The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper. Offsets are tracked by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures. In order to achieve exactly-once semantics for output of your results, your save method must be either idempotent, or an atomic transaction that saves results and offsets in your own data store.
+- *Exactly-once semantics:* The first approach uses Kafka's high level API to store consumed offsets in Zookeeper. This is traditionally the way to consume data from Kafka. While this approach (in combination with write ahead logs) can ensure zero data loss (i.e. at-least once semantics), there is a small chance some records may get consumed twice under some failures. This occurs because of inconsistencies between data reliably received by Spark Streaming and offsets tracked by Zookeeper. Hence, in this second approach, we use simple Kafka API that does not use Zookeeper. Offsets are tracked by Spark Streaming within its checkpoints. This eliminates inconsistencies between Spark Streaming and Zookeeper/Kafka, and so each record is received by Spark Streaming effectively exactly once despite failures. In order to achieve exactly-once semantics for output of your results, your output operation that saves the data to an external data store must be either idempotent, or an atomic transaction that saves results and offsets (see [Semanitcs of output operations](streaming-programming-guide.html#semantics-of-output-operations) in the main programming guide for further information).
 
 Note that one disadvantage of this approach is that it does not update offsets in Zookeeper, hence Zookeeper-based Kafka monitoring tools will not show progress. However, you can access the offsets processed by this approach in each batch and update Zookeeper yourself (see below).
 
@@ -135,23 +135,49 @@ Next, we discuss how to use this approach in your streaming application.
 
 	<div class="codetabs">
 	<div data-lang="scala" markdown="1">
-		directKafkaStream.foreachRDD { rdd => 
-		    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
-		    // offsetRanges.length = # of Kafka partitions being consumed
-		    ...
+		// hold a reference to the current offset ranges, so it can be used downstream
+		var offsetRanges = Array[OffsetRange]()
+		
+		directKafkaStream.transform { rdd =>
+		  offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
+		  rdd
+		}.map {
+                  ...
+		}.foreachRDD { rdd =>
+		  for (o <- offsetRanges) {
+		    println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
+		  }
+		  ...
 		}
 	</div>
 	<div data-lang="java" markdown="1">
-		directKafkaStream.foreachRDD(
-		    new Function<JavaPairRDD<String, String>, Void>() {
-		        @Override
-		        public Void call(JavaPairRDD<String, String> rdd) throws IOException {
-		            OffsetRange[] offsets = ((HasOffsetRanges)rdd.rdd()).offsetRanges();
-			    // offsetRanges.length() = # of Kafka partitions being consumed
-		            ...
-		            return null;
-		        }
+		// hold a reference to the current offset ranges, so it can be used downstream
+		final AtomicReference<OffsetRange[]> offsetRanges = new AtomicReference();
+		
+		directKafkaStream.transformToPair(
+		  new Function<JavaPairRDD<String, String>, JavaPairRDD<String, String>>() {
+		    @Override
+		    public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {
+		      OffsetRange[] offsets = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+		      offsetRanges.set(offsets);
+		      return rdd;
 		    }
+		  }
+		).map(
+		  ...
+		).foreachRDD(
+		  new Function<JavaPairRDD<String, String>, Void>() {
+		    @Override
+		    public Void call(JavaPairRDD<String, String> rdd) throws IOException {
+		      for (OffsetRange o : offsetRanges.get()) {
+		        System.out.println(
+		          o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset()
+		        );
+		      }
+		      ...
+		      return null;
+		    }
+		  }
 		);
 	</div>
 	<div data-lang="python" markdown="1">

external/kafka/src/test/java/org/apache/spark/streaming/kafka/JavaDirectKafkaStreamSuite.java

@@ -21,6 +21,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicReference;
 
 import scala.Tuple2;
 
@@ -70,6 +71,8 @@ public void tearDown() {
   public void testKafkaStream() throws InterruptedException {
     final String topic1 = "topic1";
     final String topic2 = "topic2";
+    // hold a reference to the current offset ranges, so it can be used downstream
+    final AtomicReference<OffsetRange[]> offsetRanges = new AtomicReference();
 
     String[] topic1data = createTopicAndSendData(topic1);
     String[] topic2data = createTopicAndSendData(topic2);
@@ -95,7 +98,8 @@ public void testKafkaStream() throws InterruptedException {
         new Function<JavaPairRDD<String, String>, JavaPairRDD<String, String>>() {
           @Override
           public JavaPairRDD<String, String> call(JavaPairRDD<String, String> rdd) throws Exception {
-            OffsetRange[] offsets = ((HasOffsetRanges)rdd.rdd()).offsetRanges();
+            OffsetRange[] offsets = ((HasOffsetRanges) rdd.rdd()).offsetRanges();
+            offsetRanges.set(offsets);
             Assert.assertEquals(offsets[0].topic(), topic1);
             return rdd;
           }
@@ -133,6 +137,11 @@ public String call(MessageAndMetadata<String, String> msgAndMd) throws Exception
           @Override
           public Void call(JavaRDD<String> rdd) throws Exception {
             result.addAll(rdd.collect());
+            for (OffsetRange o : offsetRanges.get()) {
+              System.out.println(
+                o.topic() + " " + o.partition() + " " + o.fromOffset() + " " + o.untilOffset()
+              );
+            }
             return null;
           }
         }

external/kafka/src/test/scala/org/apache/spark/streaming/kafka/DirectKafkaStreamSuite.scala

@@ -101,13 +101,21 @@ class DirectKafkaStreamSuite
 
     val allReceived = new ArrayBuffer[(String, String)]
 
-    stream.foreachRDD { rdd =>
-    // Get the offset ranges in the RDD
-      val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
+    // hold a reference to the current offset ranges, so it can be used downstream
+    var offsetRanges = Array[OffsetRange]()
+
+    stream.transform { rdd =>
+      // Get the offset ranges in the RDD
+      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
+      rdd
+    }.foreachRDD { rdd =>
+      for (o <- offsetRanges) {
+        println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
+      }
       val collected = rdd.mapPartitionsWithIndex { (i, iter) =>
       // For each partition, get size of the range in the partition,
       // and the number of items in the partition
-        val off = offsets(i)
+        val off = offsetRanges(i)
         val all = iter.toSeq
         val partSize = all.size
         val rangeSize = off.untilOffset - off.fromOffset