You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by jose-torres <gi...@git.apache.org> on 2018/01/12 22:43:33 UTC

[GitHub] spark pull request #20253: [SPARK-22908][SS] fix continuous Kafka data reade...

GitHub user jose-torres opened a pull request:

    https://github.com/apache/spark/pull/20253

    [SPARK-22908][SS] fix continuous Kafka data reader

    ## What changes were proposed in this pull request?
    
    The Kafka reader is now interruptible and can close itself.
    
    Note that this means we no longer have full failOnDataLoss support. We should not block on that, since this is causing serious test flakiness.
    
    ## How was this patch tested?
    
    I locally ran one of the ContinuousKafkaSourceSuite tests in a tight loop. Before the fix, my machine ran out of open file descriptors a few iterations in; now it works fine.


You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/jose-torres/spark fix-data-reader

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/spark/pull/20253.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 #20253
    
----
commit 0cb70f66fd64fb935a54a9098f9161368843b146
Author: Jose Torres <jo...@...>
Date:   2018-01-12T22:40:45Z

    fix data reader

----


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by jose-torres <gi...@git.apache.org>.

Github user jose-torres commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    The builds at commit 3fe76e3 were on an incomplete version of the PR, so their failures are expected.
    
    Of the 6 builds at target commit 0efc8c5, 5 passed and 1 failed. The failure was due to a stream.processAllAvailable() call in the Kafka suite; we already knew this method is inherently flake-prone for continuous processing, and had attempted to remove usages of it. I'm going to do another pass to get the rest, and then kick off 5 more attempts.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Why dont we have full failOnDataLoss support any more with this change?


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86063 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86063/testReport)** for PR 20253 at commit [`3fe76e3`](https://github.com/apache/spark/commit/3fe76e30a01698ce8732044a0c663baa277605cb).
     * This patch **fails Spark unit tests**.
     * This patch **does not merge cleanly**.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/86062/
    Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/86063/
    Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by asfgit <gi...@git.apache.org>.

Github user asfgit closed the pull request at:

    https://github.com/apache/spark/pull/20253


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Merged build finished. Test PASSed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4040 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4040/testReport)** for PR 20253 at commit [`0efc8c5`](https://github.com/apache/spark/commit/0efc8c5b7e98f3e79361f355a08fc8404d2d7d9b).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4049 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4049/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4045 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4045/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Merged build finished. Test PASSed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/86065/
    Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4051 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4051/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/86064/
    Test PASSed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4042 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4042/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    LGTM, assuming tests pass.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4045 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4045/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch **fails due to an unknown error code, -9**.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Build finished. Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4044 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4044/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86057 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86057/testReport)** for PR 20253 at commit [`0cb70f6`](https://github.com/apache/spark/commit/0cb70f66fd64fb935a54a9098f9161368843b146).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by jose-torres <gi...@git.apache.org>.

Github user jose-torres commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    The base PR was reverted, so I'm going to roll it back into here soon.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4051 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4051/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4042 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4042/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch **fails due to an unknown error code, -9**.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4047 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4047/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4041 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4041/testReport)** for PR 20253 at commit [`0efc8c5`](https://github.com/apache/spark/commit/0efc8c5b7e98f3e79361f355a08fc8404d2d7d9b).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4047 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4047/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by jose-torres <gi...@git.apache.org>.

Github user jose-torres commented on a diff in the pull request:

    https://github.com/apache/spark/pull/20253#discussion_r161359025
  
    --- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousDataSourceRDDIter.scala ---
    @@ -177,6 +176,7 @@ class DataReaderThread(
       private[continuous] var failureReason: Throwable = _
     
       override def run(): Unit = {
    +    TaskContext.setTaskContext(context)
    --- End diff --
    
    TaskContext is a ThreadLocal. We need to get it in the Kafka data reader to check if an interrupt has come.
    
    I argue that this doesn't represent an abstraction failure in the V2 API, but a workaround to the fact that Kafka consumers are badly behaved and do not respect interrupts.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on a diff in the pull request:

    https://github.com/apache/spark/pull/20253#discussion_r161901683
  
    --- Diff: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousTest.scala ---
    @@ -0,0 +1,91 @@
    +/*
    + * 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.sql.kafka010
    +
    +import scala.collection.mutable
    +
    +import org.apache.spark.SparkContext
    +import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd, SparkListenerTaskStart}
    +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
    +import org.apache.spark.sql.execution.streaming.StreamExecution
    +import org.apache.spark.sql.execution.streaming.continuous.ContinuousExecution
    +import org.apache.spark.sql.streaming.Trigger
    +import org.apache.spark.sql.test.TestSparkSession
    +
    +// Trait to configure StreamTest for kafka continuous execution tests.
    +trait KafkaContinuousTest extends KafkaSourceTest {
    +  override val defaultTrigger = Trigger.Continuous(1000)
    +  override val defaultUseV2Sink = true
    +
    +  // We need more than the default local[2] to be able to schedule all partitions simultaneously.
    +  override protected def createSparkSession = new TestSparkSession(
    +    new SparkContext(
    +      "local[10]",
    +      "continuous-stream-test-sql-context",
    +      sparkConf.set("spark.sql.testkey", "true")))
    +
    +  // In addition to setting the partitions in Kafka, we have to wait until the query has
    +  // reconfigured to the new count so the test framework can hook in properly.
    +  override protected def setTopicPartitions(
    +      topic: String, newCount: Int, query: StreamExecution) = {
    +    testUtils.addPartitions(topic, newCount)
    +    eventually(timeout(streamingTimeout)) {
    +      assert(
    +        query.lastExecution.logical.collectFirst {
    +          case DataSourceV2Relation(_, r: KafkaContinuousReader) => r
    +        }.exists(_.knownPartitions.size == newCount),
    +        s"query never reconfigured to $newCount partitions")
    +    }
    +  }
    +
    +  // Continuous processing tasks end asynchronously, so test that they actually end.
    +  private val tasksEndedListener = new SparkListener() {
    +    val activeTaskIds = mutable.Set[Long]()
    +
    +    override def onTaskStart(start: SparkListenerTaskStart): Unit = {
    +      activeTaskIds.add(start.taskInfo.taskId)
    +    }
    +
    +    override def onTaskEnd(end: SparkListenerTaskEnd): Unit = {
    +      activeTaskIds.remove(end.taskInfo.taskId)
    +    }
    +  }
    +  override def beforeEach(): Unit = {
    --- End diff --
    
    add an empty line.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4040 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4040/testReport)** for PR 20253 at commit [`0efc8c5`](https://github.com/apache/spark/commit/0efc8c5b7e98f3e79361f355a08fc8404d2d7d9b).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4043 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4043/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch **fails due to an unknown error code, -9**.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86064 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86064/testReport)** for PR 20253 at commit [`5f4f7cf`](https://github.com/apache/spark/commit/5f4f7cf6662a389abe42bfcc433d2035c5d1c35e).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Merged build finished. Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on a diff in the pull request:

    https://github.com/apache/spark/pull/20253#discussion_r161901973
  
    --- Diff: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousTest.scala ---
    @@ -0,0 +1,91 @@
    +/*
    + * 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.sql.kafka010
    +
    +import scala.collection.mutable
    +
    +import org.apache.spark.SparkContext
    +import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd, SparkListenerTaskStart}
    +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
    +import org.apache.spark.sql.execution.streaming.StreamExecution
    +import org.apache.spark.sql.execution.streaming.continuous.ContinuousExecution
    +import org.apache.spark.sql.streaming.Trigger
    +import org.apache.spark.sql.test.TestSparkSession
    +
    +// Trait to configure StreamTest for kafka continuous execution tests.
    +trait KafkaContinuousTest extends KafkaSourceTest {
    +  override val defaultTrigger = Trigger.Continuous(1000)
    +  override val defaultUseV2Sink = true
    +
    +  // We need more than the default local[2] to be able to schedule all partitions simultaneously.
    +  override protected def createSparkSession = new TestSparkSession(
    +    new SparkContext(
    +      "local[10]",
    +      "continuous-stream-test-sql-context",
    +      sparkConf.set("spark.sql.testkey", "true")))
    +
    +  // In addition to setting the partitions in Kafka, we have to wait until the query has
    +  // reconfigured to the new count so the test framework can hook in properly.
    +  override protected def setTopicPartitions(
    +      topic: String, newCount: Int, query: StreamExecution) = {
    +    testUtils.addPartitions(topic, newCount)
    +    eventually(timeout(streamingTimeout)) {
    +      assert(
    +        query.lastExecution.logical.collectFirst {
    +          case DataSourceV2Relation(_, r: KafkaContinuousReader) => r
    +        }.exists(_.knownPartitions.size == newCount),
    +        s"query never reconfigured to $newCount partitions")
    +    }
    +  }
    +
    +  // Continuous processing tasks end asynchronously, so test that they actually end.
    +  private val tasksEndedListener = new SparkListener() {
    +    val activeTaskIds = mutable.Set[Long]()
    +
    +    override def onTaskStart(start: SparkListenerTaskStart): Unit = {
    +      activeTaskIds.add(start.taskInfo.taskId)
    +    }
    +
    +    override def onTaskEnd(end: SparkListenerTaskEnd): Unit = {
    +      activeTaskIds.remove(end.taskInfo.taskId)
    +    }
    +  }
    +  override def beforeEach(): Unit = {
    +    spark.sparkContext.addSparkListener(tasksEndedListener)
    +  }
    +
    +  override def afterEach(): Unit = {
    +    eventually(timeout(streamingTimeout)) {
    +      assert(tasksEndedListener.activeTaskIds.isEmpty)
    +    }
    +    spark.sparkContext.removeSparkListener(tasksEndedListener)
    --- End diff --
    
    call super.afterEach()


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4048 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4048/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86062 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86062/testReport)** for PR 20253 at commit [`20411cb`](https://github.com/apache/spark/commit/20411cb635a82532c2496716295ba070283abf96).
     * This patch **fails Scala style tests**.
     * This patch **does not merge cleanly**.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4052 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4052/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #4052 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/4052/testReport)** for PR 20253 at commit [`4bb9c3f`](https://github.com/apache/spark/commit/4bb9c3f06f4da1c14ab24ad6a642bf831c90503f).
     * This patch passes all tests.
     * This patch merges cleanly.
     * This patch adds no public classes.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86065 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86065/testReport)** for PR 20253 at commit [`0efc8c5`](https://github.com/apache/spark/commit/0efc8c5b7e98f3e79361f355a08fc8404d2d7d9b).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] fix continuous Kafka data reader

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    **[Test build #86060 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/86060/testReport)** for PR 20253 at commit [`f575483`](https://github.com/apache/spark/commit/f5754837efbdca10398b769be07eaf53ae36f0f3).


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on a diff in the pull request:

    https://github.com/apache/spark/pull/20253#discussion_r161357488
  
    --- Diff: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaContinuousReader.scala ---
    @@ -0,0 +1,253 @@
    +/*
    + * 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.sql.kafka010
    +
    +import java.{util => ju}
    +import java.util.concurrent.TimeoutException
    +
    +import org.apache.kafka.clients.consumer.{ConsumerRecord, OffsetOutOfRangeException}
    +import org.apache.kafka.common.TopicPartition
    +
    +import org.apache.spark.TaskContext
    +import org.apache.spark.internal.Logging
    +import org.apache.spark.sql.SparkSession
    +import org.apache.spark.sql.catalyst.expressions.UnsafeRow
    +import org.apache.spark.sql.catalyst.expressions.codegen.{BufferHolder, UnsafeRowWriter}
    +import org.apache.spark.sql.catalyst.util.DateTimeUtils
    +import org.apache.spark.sql.kafka010.KafkaSource.{INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_FALSE, INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_TRUE}
    +import org.apache.spark.sql.sources.v2.reader._
    +import org.apache.spark.sql.sources.v2.streaming.reader.{ContinuousDataReader, ContinuousReader, Offset, PartitionOffset}
    +import org.apache.spark.sql.types.StructType
    +import org.apache.spark.unsafe.types.UTF8String
    +
    +/**
    + * A [[ContinuousReader]] for data from kafka.
    + *
    + * @param offsetReader  a reader used to get kafka offsets. Note that the actual data will be
    + *                      read by per-task consumers generated later.
    + * @param kafkaParams   String params for per-task Kafka consumers.
    + * @param sourceOptions The [[org.apache.spark.sql.sources.v2.DataSourceV2Options]] params which
    + *                      are not Kafka consumer params.
    + * @param metadataPath Path to a directory this reader can use for writing metadata.
    + * @param initialOffsets The Kafka offsets to start reading data at.
    + * @param failOnDataLoss Flag indicating whether reading should fail in data loss
    + *                       scenarios, where some offsets after the specified initial ones can't be
    + *                       properly read.
    + */
    +class KafkaContinuousReader(
    +    offsetReader: KafkaOffsetReader,
    +    kafkaParams: ju.Map[String, Object],
    +    sourceOptions: Map[String, String],
    +    metadataPath: String,
    +    initialOffsets: KafkaOffsetRangeLimit,
    +    failOnDataLoss: Boolean)
    +  extends ContinuousReader with SupportsScanUnsafeRow with Logging {
    +
    +  private lazy val session = SparkSession.getActiveSession.get
    +  private lazy val sc = session.sparkContext
    +
    +  // Initialized when creating read tasks. If this diverges from the partitions at the latest
    +  // offsets, we need to reconfigure.
    +  // Exposed outside this object only for unit tests.
    +  private[sql] var knownPartitions: Set[TopicPartition] = _
    +
    +  override def readSchema: StructType = KafkaOffsetReader.kafkaSchema
    +
    +  private var offset: Offset = _
    +  override def setOffset(start: ju.Optional[Offset]): Unit = {
    +    offset = start.orElse {
    +      val offsets = initialOffsets match {
    +        case EarliestOffsetRangeLimit => KafkaSourceOffset(offsetReader.fetchEarliestOffsets())
    +        case LatestOffsetRangeLimit => KafkaSourceOffset(offsetReader.fetchLatestOffsets())
    +        case SpecificOffsetRangeLimit(p) => offsetReader.fetchSpecificOffsets(p, reportDataLoss)
    +      }
    +      logInfo(s"Initial offsets: $offsets")
    +      offsets
    +    }
    +  }
    +
    +  override def getStartOffset(): Offset = offset
    +
    +  override def deserializeOffset(json: String): Offset = {
    +    KafkaSourceOffset(JsonUtils.partitionOffsets(json))
    +  }
    +
    +  override def createUnsafeRowReadTasks(): ju.List[ReadTask[UnsafeRow]] = {
    +    import scala.collection.JavaConverters._
    +
    +    val oldStartPartitionOffsets = KafkaSourceOffset.getPartitionOffsets(offset)
    +
    +    val currentPartitionSet = offsetReader.fetchEarliestOffsets().keySet
    +    val newPartitions = currentPartitionSet.diff(oldStartPartitionOffsets.keySet)
    +    val newPartitionOffsets = offsetReader.fetchEarliestOffsets(newPartitions.toSeq)
    +
    +    val deletedPartitions = oldStartPartitionOffsets.keySet.diff(currentPartitionSet)
    +    if (deletedPartitions.nonEmpty) {
    +      reportDataLoss(s"Some partitions were deleted: $deletedPartitions")
    +    }
    +
    +    val startOffsets = newPartitionOffsets ++
    +      oldStartPartitionOffsets.filterKeys(!deletedPartitions.contains(_))
    +    knownPartitions = startOffsets.keySet
    +
    +    startOffsets.toSeq.map {
    +      case (topicPartition, start) =>
    +        KafkaContinuousReadTask(
    +          topicPartition, start, kafkaParams, failOnDataLoss)
    +          .asInstanceOf[ReadTask[UnsafeRow]]
    +    }.asJava
    +  }
    +
    +  /** Stop this source and free any resources it has allocated. */
    +  def stop(): Unit = synchronized {
    +    offsetReader.close()
    +  }
    +
    +  override def commit(end: Offset): Unit = {}
    +
    +  override def mergeOffsets(offsets: Array[PartitionOffset]): Offset = {
    +    val mergedMap = offsets.map {
    +      case KafkaSourcePartitionOffset(p, o) => Map(p -> o)
    +    }.reduce(_ ++ _)
    +    KafkaSourceOffset(mergedMap)
    +  }
    +
    +  override def needsReconfiguration(): Boolean = {
    +    knownPartitions != null && offsetReader.fetchLatestOffsets().keySet != knownPartitions
    +  }
    +
    +  override def toString(): String = s"KafkaSource[$offsetReader]"
    +
    +  /**
    +   * If `failOnDataLoss` is true, this method will throw an `IllegalStateException`.
    +   * Otherwise, just log a warning.
    +   */
    +  private def reportDataLoss(message: String): Unit = {
    +    if (failOnDataLoss) {
    +      throw new IllegalStateException(message + s". $INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_TRUE")
    +    } else {
    +      logWarning(message + s". $INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_FALSE")
    +    }
    +  }
    +}
    +
    +/**
    + * A read task for continuous Kafka processing. This will be serialized and transformed into a
    + * full reader on executors.
    + *
    + * @param topicPartition The (topic, partition) pair this task is responsible for.
    + * @param startOffset The offset to start reading from within the partition.
    + * @param kafkaParams Kafka consumer params to use.
    + * @param failOnDataLoss Flag indicating whether data reader should fail if some offsets
    + *                       are skipped.
    + */
    +case class KafkaContinuousReadTask(
    +    topicPartition: TopicPartition,
    +    startOffset: Long,
    +    kafkaParams: ju.Map[String, Object],
    +    failOnDataLoss: Boolean) extends ReadTask[UnsafeRow] {
    +  override def createDataReader(): KafkaContinuousDataReader = {
    +    new KafkaContinuousDataReader(topicPartition, startOffset, kafkaParams, failOnDataLoss)
    +  }
    +}
    +
    +/**
    + * A per-task data reader for continuous Kafka processing.
    + *
    + * @param topicPartition The (topic, partition) pair this data reader is responsible for.
    + * @param startOffset The offset to start reading from within the partition.
    + * @param kafkaParams Kafka consumer params to use.
    + * @param failOnDataLoss Flag indicating whether data reader should fail if some offsets
    + *                       are skipped.
    + */
    +class KafkaContinuousDataReader(
    +    topicPartition: TopicPartition,
    +    startOffset: Long,
    +    kafkaParams: ju.Map[String, Object],
    +    failOnDataLoss: Boolean) extends ContinuousDataReader[UnsafeRow] {
    +  private val topic = topicPartition.topic
    +  private val kafkaPartition = topicPartition.partition
    +  private val consumer = CachedKafkaConsumer.createUncached(topic, kafkaPartition, kafkaParams)
    +
    +  private val sharedRow = new UnsafeRow(7)
    +  private val bufferHolder = new BufferHolder(sharedRow)
    +  private val rowWriter = new UnsafeRowWriter(bufferHolder, 7)
    +
    +  private var nextKafkaOffset = startOffset
    +  private var currentRecord: ConsumerRecord[Array[Byte], Array[Byte]] = _
    +
    +  override def next(): Boolean = {
    +    var r: ConsumerRecord[Array[Byte], Array[Byte]] = null
    +    while (r == null) {
    +      if (TaskContext.get().isInterrupted() || TaskContext.get().isCompleted()) return false
    +      // Our consumer.get is not interruptible, so we have to set a low poll timeout, leaving
    +      // interrupt points to end the query rather than waiting for new data that might never come.
    +      try {
    +        r = consumer.get(
    +          nextKafkaOffset,
    +          untilOffset = Long.MaxValue,
    +          pollTimeoutMs = 1000,
    +          failOnDataLoss)
    +      } catch {
    +        // We didn't read within the timeout. We're supposed to block indefinitely for new data, so
    +        // swallow and ignore this.
    +        case _: TimeoutException =>
    +        // This is a failOnDataLoss exception. Retry if nextKafkaOffset is within the data range,
    --- End diff --
    
    add an empty line to distinguish that comment below this belongs to next case


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Merged build finished. Test FAILed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #20253: [SPARK-22908][SS] Roll forward continuous processing Kaf...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/20253
  
    Merged build finished. Test PASSed.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #20253: [SPARK-22908][SS] Roll forward continuous process...

Posted by tdas <gi...@git.apache.org>.

Github user tdas commented on a diff in the pull request:

    https://github.com/apache/spark/pull/20253#discussion_r161357974
  
    --- Diff: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaContinuousReader.scala ---
    @@ -0,0 +1,253 @@
    +/*
    + * 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.sql.kafka010
    +
    +import java.{util => ju}
    +import java.util.concurrent.TimeoutException
    +
    +import org.apache.kafka.clients.consumer.{ConsumerRecord, OffsetOutOfRangeException}
    +import org.apache.kafka.common.TopicPartition
    +
    +import org.apache.spark.TaskContext
    +import org.apache.spark.internal.Logging
    +import org.apache.spark.sql.SparkSession
    +import org.apache.spark.sql.catalyst.expressions.UnsafeRow
    +import org.apache.spark.sql.catalyst.expressions.codegen.{BufferHolder, UnsafeRowWriter}
    +import org.apache.spark.sql.catalyst.util.DateTimeUtils
    +import org.apache.spark.sql.kafka010.KafkaSource.{INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_FALSE, INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_TRUE}
    +import org.apache.spark.sql.sources.v2.reader._
    +import org.apache.spark.sql.sources.v2.streaming.reader.{ContinuousDataReader, ContinuousReader, Offset, PartitionOffset}
    +import org.apache.spark.sql.types.StructType
    +import org.apache.spark.unsafe.types.UTF8String
    +
    +/**
    + * A [[ContinuousReader]] for data from kafka.
    + *
    + * @param offsetReader  a reader used to get kafka offsets. Note that the actual data will be
    + *                      read by per-task consumers generated later.
    + * @param kafkaParams   String params for per-task Kafka consumers.
    + * @param sourceOptions The [[org.apache.spark.sql.sources.v2.DataSourceV2Options]] params which
    + *                      are not Kafka consumer params.
    + * @param metadataPath Path to a directory this reader can use for writing metadata.
    + * @param initialOffsets The Kafka offsets to start reading data at.
    + * @param failOnDataLoss Flag indicating whether reading should fail in data loss
    + *                       scenarios, where some offsets after the specified initial ones can't be
    + *                       properly read.
    + */
    +class KafkaContinuousReader(
    +    offsetReader: KafkaOffsetReader,
    +    kafkaParams: ju.Map[String, Object],
    +    sourceOptions: Map[String, String],
    +    metadataPath: String,
    +    initialOffsets: KafkaOffsetRangeLimit,
    +    failOnDataLoss: Boolean)
    +  extends ContinuousReader with SupportsScanUnsafeRow with Logging {
    +
    +  private lazy val session = SparkSession.getActiveSession.get
    +  private lazy val sc = session.sparkContext
    +
    +  // Initialized when creating read tasks. If this diverges from the partitions at the latest
    +  // offsets, we need to reconfigure.
    +  // Exposed outside this object only for unit tests.
    +  private[sql] var knownPartitions: Set[TopicPartition] = _
    +
    +  override def readSchema: StructType = KafkaOffsetReader.kafkaSchema
    +
    +  private var offset: Offset = _
    +  override def setOffset(start: ju.Optional[Offset]): Unit = {
    +    offset = start.orElse {
    +      val offsets = initialOffsets match {
    +        case EarliestOffsetRangeLimit => KafkaSourceOffset(offsetReader.fetchEarliestOffsets())
    +        case LatestOffsetRangeLimit => KafkaSourceOffset(offsetReader.fetchLatestOffsets())
    +        case SpecificOffsetRangeLimit(p) => offsetReader.fetchSpecificOffsets(p, reportDataLoss)
    +      }
    +      logInfo(s"Initial offsets: $offsets")
    +      offsets
    +    }
    +  }
    +
    +  override def getStartOffset(): Offset = offset
    +
    +  override def deserializeOffset(json: String): Offset = {
    +    KafkaSourceOffset(JsonUtils.partitionOffsets(json))
    +  }
    +
    +  override def createUnsafeRowReadTasks(): ju.List[ReadTask[UnsafeRow]] = {
    +    import scala.collection.JavaConverters._
    +
    +    val oldStartPartitionOffsets = KafkaSourceOffset.getPartitionOffsets(offset)
    +
    +    val currentPartitionSet = offsetReader.fetchEarliestOffsets().keySet
    +    val newPartitions = currentPartitionSet.diff(oldStartPartitionOffsets.keySet)
    +    val newPartitionOffsets = offsetReader.fetchEarliestOffsets(newPartitions.toSeq)
    +
    +    val deletedPartitions = oldStartPartitionOffsets.keySet.diff(currentPartitionSet)
    +    if (deletedPartitions.nonEmpty) {
    +      reportDataLoss(s"Some partitions were deleted: $deletedPartitions")
    +    }
    +
    +    val startOffsets = newPartitionOffsets ++
    +      oldStartPartitionOffsets.filterKeys(!deletedPartitions.contains(_))
    +    knownPartitions = startOffsets.keySet
    +
    +    startOffsets.toSeq.map {
    +      case (topicPartition, start) =>
    +        KafkaContinuousReadTask(
    +          topicPartition, start, kafkaParams, failOnDataLoss)
    +          .asInstanceOf[ReadTask[UnsafeRow]]
    +    }.asJava
    +  }
    +
    +  /** Stop this source and free any resources it has allocated. */
    +  def stop(): Unit = synchronized {
    +    offsetReader.close()
    +  }
    +
    +  override def commit(end: Offset): Unit = {}
    +
    +  override def mergeOffsets(offsets: Array[PartitionOffset]): Offset = {
    +    val mergedMap = offsets.map {
    +      case KafkaSourcePartitionOffset(p, o) => Map(p -> o)
    +    }.reduce(_ ++ _)
    +    KafkaSourceOffset(mergedMap)
    +  }
    +
    +  override def needsReconfiguration(): Boolean = {
    +    knownPartitions != null && offsetReader.fetchLatestOffsets().keySet != knownPartitions
    +  }
    +
    +  override def toString(): String = s"KafkaSource[$offsetReader]"
    +
    +  /**
    +   * If `failOnDataLoss` is true, this method will throw an `IllegalStateException`.
    +   * Otherwise, just log a warning.
    +   */
    +  private def reportDataLoss(message: String): Unit = {
    +    if (failOnDataLoss) {
    +      throw new IllegalStateException(message + s". $INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_TRUE")
    +    } else {
    +      logWarning(message + s". $INSTRUCTION_FOR_FAIL_ON_DATA_LOSS_FALSE")
    +    }
    +  }
    +}
    +
    +/**
    + * A read task for continuous Kafka processing. This will be serialized and transformed into a
    + * full reader on executors.
    + *
    + * @param topicPartition The (topic, partition) pair this task is responsible for.
    + * @param startOffset The offset to start reading from within the partition.
    + * @param kafkaParams Kafka consumer params to use.
    + * @param failOnDataLoss Flag indicating whether data reader should fail if some offsets
    + *                       are skipped.
    + */
    +case class KafkaContinuousReadTask(
    +    topicPartition: TopicPartition,
    +    startOffset: Long,
    +    kafkaParams: ju.Map[String, Object],
    +    failOnDataLoss: Boolean) extends ReadTask[UnsafeRow] {
    +  override def createDataReader(): KafkaContinuousDataReader = {
    +    new KafkaContinuousDataReader(topicPartition, startOffset, kafkaParams, failOnDataLoss)
    +  }
    +}
    +
    +/**
    + * A per-task data reader for continuous Kafka processing.
    + *
    + * @param topicPartition The (topic, partition) pair this data reader is responsible for.
    + * @param startOffset The offset to start reading from within the partition.
    + * @param kafkaParams Kafka consumer params to use.
    + * @param failOnDataLoss Flag indicating whether data reader should fail if some offsets
    + *                       are skipped.
    + */
    +class KafkaContinuousDataReader(
    +    topicPartition: TopicPartition,
    +    startOffset: Long,
    +    kafkaParams: ju.Map[String, Object],
    +    failOnDataLoss: Boolean) extends ContinuousDataReader[UnsafeRow] {
    +  private val topic = topicPartition.topic
    +  private val kafkaPartition = topicPartition.partition
    +  private val consumer = CachedKafkaConsumer.createUncached(topic, kafkaPartition, kafkaParams)
    +
    +  private val sharedRow = new UnsafeRow(7)
    +  private val bufferHolder = new BufferHolder(sharedRow)
    +  private val rowWriter = new UnsafeRowWriter(bufferHolder, 7)
    +
    +  private var nextKafkaOffset = startOffset
    +  private var currentRecord: ConsumerRecord[Array[Byte], Array[Byte]] = _
    +
    +  override def next(): Boolean = {
    +    var r: ConsumerRecord[Array[Byte], Array[Byte]] = null
    +    while (r == null) {
    +      if (TaskContext.get().isInterrupted() || TaskContext.get().isCompleted()) return false
    +      // Our consumer.get is not interruptible, so we have to set a low poll timeout, leaving
    +      // interrupt points to end the query rather than waiting for new data that might never come.
    +      try {
    +        r = consumer.get(
    +          nextKafkaOffset,
    +          untilOffset = Long.MaxValue,
    +          pollTimeoutMs = 1000,
    --- End diff --
    
    Make this configurable. In fact, we already have kafka option "kafkaConsumer.pollTimeoutMs" which means roughly this (search for it here https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html). It has a low default value of 512ms. This should be good.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org