[SPARK-4964] javafication of interfaces

external/kafka/src/main/java/org/apache/spark/streaming/kafka/HasOffsetRanges.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.streaming.kafka;
+
+/** Something that has a collection of OffsetRanges */
+public interface HasOffsetRanges {
+  /** array of OffsetRanges */
+  public OffsetRange[] offsetRanges();
+}

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/Leader.scala → external/kafka/src/main/java/org/apache/spark/streaming/kafka/Leader.java

@@ -15,54 +15,19 @@
  * limitations under the License.
  */
 
-package org.apache.spark.streaming.kafka
-
-import kafka.common.TopicAndPartition
+package org.apache.spark.streaming.kafka;
 
 /** Host info for the leader of a Kafka TopicAndPartition */
-
-trait Leader {
-    /** kafka topic name */
-  def topic: String
+public interface Leader {
+  /** kafka topic name */
+  public String topic();
 
   /** kafka partition id */
-  def partition: Int
+  public int partition();
 
   /** kafka hostname */
-  def host: String
+  public String host();
 
   /** kafka host's port */
-  def port: Int
-}
-
-private class LeaderImpl(
-  override val topic: String,
-  override val partition: Int,
-  override val host: String,
-  override val port: Int
-) extends Leader
-
-object Leader {
-  def create(
-    topic: String,
-    partition: Int,
-    host: String,
-    port: Int): Leader =
-    new LeaderImpl(
-      topic,
-      partition,
-      host,
-      port)
-
-    def create(
-    topicAndPartition: TopicAndPartition,
-    host: String,
-    port: Int): Leader =
-    new LeaderImpl(
-      topicAndPartition.topic,
-      topicAndPartition.partition,
-      host,
-      port)
-
+  public int port();
 }
-

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/OffsetRange.scala → external/kafka/src/main/java/org/apache/spark/streaming/kafka/OffsetRange.java

@@ -15,57 +15,19 @@
  * limitations under the License.
  */
 
-package org.apache.spark.streaming.kafka
-
-import kafka.common.TopicAndPartition
+package org.apache.spark.streaming.kafka;
 
 /** Represents a range of offsets from a single Kafka TopicAndPartition */
-trait OffsetRange {
+public interface OffsetRange {
   /** kafka topic name */
-  def topic: String
+  public String topic();
 
   /** kafka partition id */
-  def partition: Int
+  public int partition();
 
   /** inclusive starting offset */
-  def fromOffset: Long
+  public long fromOffset();
 
   /** exclusive ending offset */
-  def untilOffset: Long
-}
-
-/** Something that has a collection of OffsetRanges */
-trait HasOffsetRanges {
-  def offsetRanges: Array[OffsetRange]
-}
-
-private class OffsetRangeImpl(
-  override val topic: String,
-  override val partition: Int,
-  override val fromOffset: Long,
-  override val untilOffset: Long
-) extends OffsetRange
-
-object OffsetRange {
-  def create(
-    topic: String,
-    partition: Int,
-    fromOffset: Long,
-    untilOffset: Long): OffsetRange =
-    new OffsetRangeImpl(
-      topic,
-      partition,
-      fromOffset,
-      untilOffset)
-
-  def create(
-    topicAndPartition: TopicAndPartition,
-    fromOffset: Long,
-    untilOffset: Long): OffsetRange =
-    new OffsetRangeImpl(
-      topicAndPartition.topic,
-      topicAndPartition.partition,
-      fromOffset,
-      untilOffset)
-
+  public long untilOffset();
 }

external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaUtils.scala

@@ -330,4 +330,75 @@ object KafkaUtils {
       ok => ok
     )
   }
+
+  private class OffsetRangeImpl(
+      override val topic: String,
+      override val partition: Int,
+      override val fromOffset: Long,
+      override val untilOffset: Long) extends OffsetRange
+
+  /**
+   * Behaviorless container for a range of offsets from a single Kafka TopicAndPartition
+   * @param topic kafka topic name
+   * @param partition kafka partition id
+   * @param fromOffset inclusive starting offset
+   * @param untilOffset exclusive ending offset
+ */
+  @Experimental
+  def createOffsetRange(
+      topic: String,
+      partition: Int,
+      fromOffset: Long,
+      untilOffset: Long): OffsetRange =
+    new OffsetRangeImpl(topic, partition, fromOffset, untilOffset)
+
+  /**
+   * Behaviorless container for a range of offsets from a single Kafka TopicAndPartition
+   * @param topicAndPartition kafka TopicAndPartition
+   * @param fromOffset inclusive starting offset
+   * @param untilOffset exclusive ending offset
+ */
+  @Experimental
+  def createOffsetRange(
+      topicAndPartition: TopicAndPartition,
+      fromOffset: Long,
+      untilOffset: Long): OffsetRange =
+    new OffsetRangeImpl(
+      topicAndPartition.topic, topicAndPartition.partition, fromOffset, untilOffset)
+
+  private class LeaderImpl(
+      override val topic: String,
+      override val partition: Int,
+      override val host: String,
+      override val port: Int) extends Leader
+
+  /**
+   * Behaviorless container of host info for the leader of a Kafka TopicAndPartition
+   * @param topic kafka topic name
+   * @param partition kafka partition id
+   * @param host kafka hostname
+   * @param port kafka host's port
+   */
+  @Experimental
+  def createLeader(topic: String, partition: Int, host: String, port: Int): Leader =
+    new LeaderImpl(topic,partition,
+      host,
+      port)
+
+  /**
+   * Behaviorless container of host info for the leader of a Kafka TopicAndPartition
+   * @param topicAndPartition kafka TopicAndPartition
+   * @param host kafka hostname
+   * @param port kafka host's port
+   */
+  @Experimental
+  def createLeader(
+    topicAndPartition: TopicAndPartition,
+    host: String,
+    port: Int): Leader =
+    new LeaderImpl(
+      topicAndPartition.topic,
+      topicAndPartition.partition,
+      host,
+      port)
 }