You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by zsxwing <gi...@git.apache.org> on 2015/11/02 22:41:21 UTC
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
GitHub user zsxwing opened a pull request:
https://github.com/apache/spark/pull/9420
[SPARK-11462][Streaming]Add JavaStreamingListener
Currently, StreamingListener is not Java friendly because it exposes some Scala collections to Java users directly, such as Option, Map.
This PR added a Java version of StreamingListener and a bunch of Java friendly classes for Java users.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zsxwing/spark java-streaming-listener
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/9420.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #9420
----
commit 2bcb84d6aaac78e5eb9c168483741c1e5111f413
Author: zsxwing <zs...@gmail.com>
Date: 2015-11-02T18:26:43Z
Add JavaStreamingListener
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153169919
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/44835/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/9420#discussion_r44333044
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala ---
@@ -0,0 +1,189 @@
+/*
+ * 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.api.java
+
+import org.apache.spark.annotation.DeveloperApi
+import org.apache.spark.streaming.Time
+import org.apache.spark.streaming.scheduler._
+
+/**
+ * A listener interface for receiving information about an ongoing streaming computation.
+ */
+@DeveloperApi
+abstract class JavaStreamingListener {
+
+ /** Called when a receiver has been started */
+ def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = { }
+
+ /** Called when a receiver has reported an error */
+ def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = { }
+
+ /** Called when a receiver has been stopped */
+ def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = { }
+
+ /** Called when a batch of jobs has been submitted for processing. */
+ def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = { }
+
+ /** Called when processing of a batch of jobs has started. */
+ def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = { }
+
+ /** Called when processing of a batch of jobs has completed. */
+ def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = { }
+
+ /** Called when processing of a job of a batch has started. */
+ def onOutputOperationStarted(
+ outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = { }
+
+ /** Called when processing of a job of a batch has completed. */
+ def onOutputOperationCompleted(
+ outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = { }
+}
+
+/**
+ * :: DeveloperApi ::
+ * Base trait for events related to JavaStreamingListener
+ */
+@DeveloperApi
+sealed trait JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchSubmitted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchCompleted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchStarted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerOutputOperationStarted(val outputOperationInfo: JavaOutputOperationInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerOutputOperationCompleted(
+ val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverStarted(val receiverInfo: JavaReceiverInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverError(val receiverInfo: JavaReceiverInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverStopped(val receiverInfo: JavaReceiverInfo)
--- End diff --
The JavaSparkListener uses the same object types as SparkListener in the method. You seem to have defined additional Java classes, why? To avoid options?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153911924
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/9420
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153911938
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by djalova <gi...@git.apache.org>.
Github user djalova commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153909707
Can you test this again please?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153912823
**[Test build #45068 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/45068/consoleFull)** for PR 9420 at commit [`2bcb84d`](https://github.com/apache/spark/commit/2bcb84d6aaac78e5eb9c168483741c1e5111f413).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155255001
All right. Merging this to master and 1.6. Thanks @zsxwing for doing this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155216274
**[Test build #45426 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/45426/consoleFull)** for PR 9420 at commit [`c21e786`](https://github.com/apache/spark/commit/c21e786c72b5b89a22c6a3f2237ee33e09e55e0b).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153923931
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/45068/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/9420#discussion_r44345630
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.api.java
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.streaming.scheduler._
+
+/**
+ * A wrapper to convert a [[JavaStreamingListener]] to a [[StreamingListener]].
+ */
+private[streaming] class JavaStreamingListenerWrapper(javaStreamingListener: JavaStreamingListener)
+ extends StreamingListener {
+
+ private def toJavaReceiverInfo(receiverInfo: ReceiverInfo): JavaReceiverInfo = {
+ JavaReceiverInfo(
+ receiverInfo.streamId,
+ receiverInfo.name,
+ receiverInfo.active,
+ receiverInfo.location,
+ receiverInfo.lastErrorMessage,
+ receiverInfo.lastError,
+ receiverInfo.lastErrorTime
+ )
+ }
+
+ private def toJavaStreamInputInfo(streamInputInfo: StreamInputInfo): JavaStreamInputInfo = {
+ JavaStreamInputInfo(
+ streamInputInfo.inputStreamId,
+ streamInputInfo.numRecords: Long,
+ streamInputInfo.metadata.asJava,
+ streamInputInfo.metadataDescription.orNull
+ )
+ }
+
+ private def toJavaOutputOperationInfo(
+ outputOperationInfo: OutputOperationInfo): JavaOutputOperationInfo = {
+ JavaOutputOperationInfo(
+ outputOperationInfo.batchTime,
+ outputOperationInfo.id,
+ outputOperationInfo.name,
+ outputOperationInfo.description: String,
+ outputOperationInfo.startTime.getOrElse(-1),
+ outputOperationInfo.endTime.getOrElse(-1),
+ outputOperationInfo.failureReason.orNull
+ )
+ }
+
+ private def toJavaBatchInfo(batchInfo: BatchInfo): JavaBatchInfo = {
+ JavaBatchInfo(
+ batchInfo.batchTime,
+ batchInfo.streamIdToInputInfo.mapValues(toJavaStreamInputInfo(_)).asJava,
+ batchInfo.submissionTime,
+ batchInfo.processingStartTime.getOrElse(-1),
--- End diff --
> If the Java API is going to use -1 as "not set" (equivalent to Python), then would the python being using -1 as well?
Yes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153923927
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155214254
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155214284
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155228507
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/9420#discussion_r44334549
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListener.scala ---
@@ -0,0 +1,189 @@
+/*
+ * 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.api.java
+
+import org.apache.spark.annotation.DeveloperApi
+import org.apache.spark.streaming.Time
+import org.apache.spark.streaming.scheduler._
+
+/**
+ * A listener interface for receiving information about an ongoing streaming computation.
+ */
+@DeveloperApi
+abstract class JavaStreamingListener {
+
+ /** Called when a receiver has been started */
+ def onReceiverStarted(receiverStarted: JavaStreamingListenerReceiverStarted): Unit = { }
+
+ /** Called when a receiver has reported an error */
+ def onReceiverError(receiverError: JavaStreamingListenerReceiverError): Unit = { }
+
+ /** Called when a receiver has been stopped */
+ def onReceiverStopped(receiverStopped: JavaStreamingListenerReceiverStopped): Unit = { }
+
+ /** Called when a batch of jobs has been submitted for processing. */
+ def onBatchSubmitted(batchSubmitted: JavaStreamingListenerBatchSubmitted): Unit = { }
+
+ /** Called when processing of a batch of jobs has started. */
+ def onBatchStarted(batchStarted: JavaStreamingListenerBatchStarted): Unit = { }
+
+ /** Called when processing of a batch of jobs has completed. */
+ def onBatchCompleted(batchCompleted: JavaStreamingListenerBatchCompleted): Unit = { }
+
+ /** Called when processing of a job of a batch has started. */
+ def onOutputOperationStarted(
+ outputOperationStarted: JavaStreamingListenerOutputOperationStarted): Unit = { }
+
+ /** Called when processing of a job of a batch has completed. */
+ def onOutputOperationCompleted(
+ outputOperationCompleted: JavaStreamingListenerOutputOperationCompleted): Unit = { }
+}
+
+/**
+ * :: DeveloperApi ::
+ * Base trait for events related to JavaStreamingListener
+ */
+@DeveloperApi
+sealed trait JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchSubmitted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchCompleted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerBatchStarted(val batchInfo: JavaBatchInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerOutputOperationStarted(val outputOperationInfo: JavaOutputOperationInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerOutputOperationCompleted(
+ val outputOperationInfo: JavaOutputOperationInfo) extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverStarted(val receiverInfo: JavaReceiverInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverError(val receiverInfo: JavaReceiverInfo)
+ extends JavaStreamingListenerEvent
+
+@DeveloperApi
+class JavaStreamingListenerReceiverStopped(val receiverInfo: JavaReceiverInfo)
--- End diff --
`JavaSparkListener` is not java friendly. So I don't follow it. IMO, we should not use any Scala classes in the public Java API.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153166192
Merged build triggered.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-155228371
**[Test build #45426 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/45426/consoleFull)** for PR 9420 at commit [`c21e786`](https://github.com/apache/spark/commit/c21e786c72b5b89a22c6a3f2237ee33e09e55e0b).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-154514213
This would have to be reviewed by @tdas
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153166227
Merged build started.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by djalova <gi...@git.apache.org>.
Github user djalova commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-154488931
Will this be merged soon?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153169915
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/9420#discussion_r44344771
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaStreamingListenerWrapper.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.api.java
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.streaming.scheduler._
+
+/**
+ * A wrapper to convert a [[JavaStreamingListener]] to a [[StreamingListener]].
+ */
+private[streaming] class JavaStreamingListenerWrapper(javaStreamingListener: JavaStreamingListener)
+ extends StreamingListener {
+
+ private def toJavaReceiverInfo(receiverInfo: ReceiverInfo): JavaReceiverInfo = {
+ JavaReceiverInfo(
+ receiverInfo.streamId,
+ receiverInfo.name,
+ receiverInfo.active,
+ receiverInfo.location,
+ receiverInfo.lastErrorMessage,
+ receiverInfo.lastError,
+ receiverInfo.lastErrorTime
+ )
+ }
+
+ private def toJavaStreamInputInfo(streamInputInfo: StreamInputInfo): JavaStreamInputInfo = {
+ JavaStreamInputInfo(
+ streamInputInfo.inputStreamId,
+ streamInputInfo.numRecords: Long,
+ streamInputInfo.metadata.asJava,
+ streamInputInfo.metadataDescription.orNull
+ )
+ }
+
+ private def toJavaOutputOperationInfo(
+ outputOperationInfo: OutputOperationInfo): JavaOutputOperationInfo = {
+ JavaOutputOperationInfo(
+ outputOperationInfo.batchTime,
+ outputOperationInfo.id,
+ outputOperationInfo.name,
+ outputOperationInfo.description: String,
+ outputOperationInfo.startTime.getOrElse(-1),
+ outputOperationInfo.endTime.getOrElse(-1),
+ outputOperationInfo.failureReason.orNull
+ )
+ }
+
+ private def toJavaBatchInfo(batchInfo: BatchInfo): JavaBatchInfo = {
+ JavaBatchInfo(
+ batchInfo.batchTime,
+ batchInfo.streamIdToInputInfo.mapValues(toJavaStreamInputInfo(_)).asJava,
+ batchInfo.submissionTime,
+ batchInfo.processingStartTime.getOrElse(-1),
--- End diff --
If the Java API is going to use -1 as "not set" (equivalent to Python), then would the python being using -1 as well?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153923808
**[Test build #45068 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/45068/consoleFull)** for PR 9420 at commit [`2bcb84d`](https://github.com/apache/spark/commit/2bcb84d6aaac78e5eb9c168483741c1e5111f413).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:\n * `abstract class JavaStreamingListener `\n * `class JavaStreamingListenerBatchSubmitted(val batchInfo: JavaBatchInfo)`\n * `class JavaStreamingListenerBatchCompleted(val batchInfo: JavaBatchInfo)`\n * `class JavaStreamingListenerBatchStarted(val batchInfo: JavaBatchInfo)`\n * `class JavaStreamingListenerOutputOperationStarted(val outputOperationInfo: JavaOutputOperationInfo)`\n * `class JavaStreamingListenerOutputOperationCompleted(`\n * `class JavaStreamingListenerReceiverStarted(val receiverInfo: JavaReceiverInfo)`\n * `class JavaStreamingListenerReceiverError(val receiverInfo: JavaReceiverInfo)`\n * `class JavaStreamingListenerReceiverStopped(val receiverInfo: JavaReceiverInfo)`\n * `case class JavaBatchInfo(`\n * `case class JavaStreamInputInfo(`\n * `case class JavaReceiverInfo(`\n * `case class JavaOutputOperationInfo(`\n
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-11462][Streaming]Add JavaStreamingListe...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/9420#issuecomment-153910446
retest this please
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org