[SPARK-25644][SS]Fix java foreachBatch in DataStreamWriter

sql/core/src/main/scala/org/apache/spark/sql/streaming/DataStreamWriter.scala

@@ -380,7 +380,7 @@ final class DataStreamWriter[T] private[sql](ds: Dataset[T]) {
    * @since 2.4.0
    */
   @InterfaceStability.Evolving
-  def foreachBatch(function: VoidFunction2[Dataset[T], Long]): DataStreamWriter[T] = {
+  def foreachBatch(function: VoidFunction2[Dataset[T], java.lang.Long]): DataStreamWriter[T] = {
     foreachBatch((batchDs: Dataset[T], batchId: Long) => function.call(batchDs, batchId))
   }
 

sql/core/src/test/java/test/org/apache/spark/sql/streaming/JavaDataStreamReaderWriterSuite.java

@@ -0,0 +1,89 @@
+/*
+* 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 test.org.apache.spark.sql.streaming;
+
+import java.io.File;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.spark.api.java.function.VoidFunction2;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.ForeachWriter;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.streaming.StreamingQuery;
+import org.apache.spark.sql.test.TestSparkSession;
+import org.apache.spark.util.Utils;
+
+public class JavaDataStreamReaderWriterSuite {
+  private SparkSession spark;
+  private String input;
+
+  @Before
+  public void setUp() {
+    spark = new TestSparkSession();
+    input = Utils.createTempDir(System.getProperty("java.io.tmpdir"), "input").toString();
+  }
+
+  @After
+  public void tearDown() {
+    Utils.deleteRecursively(new File(input));
+    spark.stop();
+    spark = null;
+  }
+
+  @Test
+  public void testForeachBatchAPI() {
+    StreamingQuery query = spark
+        .readStream()
+        .textFile(input)
+        .writeStream()
+        .foreachBatch(new VoidFunction2<Dataset<String>, Long>() {
+          @Override
+          public void call(Dataset<String> v1, Long v2) throws Exception {
+          }
+        })
+        .start();
+    query.stop();
+  }
+
+  @Test
+  public void testForeachAPI() {
+    StreamingQuery query = spark
+        .readStream()
+        .textFile(input)
+        .writeStream()
+        .foreach(new ForeachWriter<String>() {
+          @Override
+          public boolean open(long partitionId, long epochId) {
+            return true;
+          }
+
+          @Override
+          public void process(String value) {
+          }
+
+          @Override
+          public void close(Throwable errorOrNull) {
+          }
+        })
+        .start();
+    query.stop();
+  }
+}