[SPARK-17834][SQL]Fetch the earliest offsets manually in KafkaSource instead of counting on KafkaConsumer #15397

zsxwing · 2016-10-07T23:51:04Z

What changes were proposed in this pull request?

Because KafkaConsumer.poll(0) may update the partition offsets, this PR just calls seekToBeginning to manually set the earliest offsets for the KafkaSource initial offsets.

How was this patch tested?

Existing tests.

…g on KafkaConsumer

zsxwing · 2016-10-07T23:51:58Z

/cc @tdas @koeninger

SparkQA · 2016-10-08T00:24:06Z

Test build #66543 has finished for PR 15397 at commit 95a0c96.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

koeninger · 2016-10-08T01:05:03Z

How is this going to work with assign? It seems like it's just avoiding the problem, not fixing it.

zsxwing · 2016-10-10T04:40:28Z

How is this going to work with assign? It seems like it's just avoiding the problem, not fixing it.

We can seek to the offsets provided by the user.

koeninger · 2016-10-10T14:25:52Z

Look at the poll/seek implementation in the DStream's subscribe and
subscribe pattern when user offsets are provided, i.e. the problem that
triggered this ticket to begin with. You're going to have to solve the same
problem there with the structured stream, unless the structured stream
somehow wants to limit assigning specific partitions only to the assign
strategy, which eliminates lots of valid use cases.

On Sun, Oct 9, 2016 at 11:41 PM, Shixiong Zhu notifications@github.com
wrote:

How is this going to work with assign? It seems like it's just avoiding
the problem, not fixing it.

We can seek to the offsets provided by the user.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
#15397 (comment), or mute
the thread
https://github.com/notifications/unsubscribe-auth/AAGAB1CurDhXuJr-NxZrbI_7wG8O_2l7ks5qycHzgaJpZM4KRjTS
.

zsxwing · 2016-10-12T22:06:49Z

@koeninger sorry for the delay. Right now KafkaSource doesn't support external group id, so we don't need to concern about how to fetching committed offsets. Any other cases that I'm missing?

SparkQA · 2016-10-12T22:36:33Z

Test build #66849 has finished for PR 15397 at commit 9578555.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

koeninger · 2016-10-12T22:43:56Z

My main point is that whoever implements SPARK-17812 is going to have to deal with the issue shown in SPARK-17782, which means much of this patch is going to need to be changed anyway.

But It's not just about external group id. Committed offsets would actually make the issue in SPARK-17782 less of a problem, because they would take precedence over auto.offset.reset

zsxwing · 2016-10-13T18:48:00Z

My main point is that whoever implements SPARK-17812 is going to have to deal with the issue shown in SPARK-17782, which means much of this patch is going to need to be changed anyway.

@koeninger I agreed that this patch will be changed. However, this PR does fix a known issue for the current supported features and there is not user facing changes. Considering 2.0.2 may come out soon and I don't think SPARK-17812 will be done soon, I would like to merge this to fix issues for 2.0.2. What do you think?

koeninger

Did another once over, couple more minor things.

If the plan is to wait for SPARK-17812 to fix up the other stuff I was concerned about, that's ok with me, but I really hope it doesn't slip past another release. To reiterate, I'm fine with doing that work.

koeninger · 2016-10-13T19:06:02Z

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala

@@ -256,8 +269,6 @@ private[kafka010] case class KafkaSource(
   */
  private def fetchNewPartitionEarliestOffsets(
      newPartitions: Seq[TopicPartition]): Map[TopicPartition, Long] = withRetriesWithoutInterrupt {
-    // Make sure `KafkaConsumer.poll` won't be interrupted (KAFKA-1894)
-    assert(Thread.currentThread().isInstanceOf[StreamExecutionThread])
    // Poll to get the latest assigned partitions
    consumer.poll(0)
    val partitions = consumer.assignment()


Is there a reason not to pause all partitions here as well?

koeninger · 2016-10-13T19:06:07Z

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala

@@ -270,7 +281,7 @@ private[kafka010] case class KafkaSource(
      // So we need to ignore them
      partitions.contains(p)
    }.map(p => p -> consumer.position(p)).toMap
-    logDebug(s"Got offsets for new partitions: $partitionToOffsets")
+    logDebug(s"Got earliest offsets for new partitions: $partitionToOffsets")
    partitionToOffsets


nit: different variable name partitionToOffsets vs partitionOffsets for what is essentially the same thing

SparkQA · 2016-10-13T20:00:43Z

Test build #66908 has finished for PR 15397 at commit 7986f18.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

koeninger · 2016-10-13T20:10:12Z

LGTM, thanks for talking it through

zsxwing · 2016-10-13T20:31:31Z

Thanks! Merging to master and 2.0.

… instead of counting on KafkaConsumer ## What changes were proposed in this pull request? Because `KafkaConsumer.poll(0)` may update the partition offsets, this PR just calls `seekToBeginning` to manually set the earliest offsets for the KafkaSource initial offsets. ## How was this patch tested? Existing tests. Author: Shixiong Zhu <shixiong@databricks.com> Closes #15397 from zsxwing/SPARK-17834. (cherry picked from commit 08eac35) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

… instead of counting on KafkaConsumer ## What changes were proposed in this pull request? Because `KafkaConsumer.poll(0)` may update the partition offsets, this PR just calls `seekToBeginning` to manually set the earliest offsets for the KafkaSource initial offsets. ## How was this patch tested? Existing tests. Author: Shixiong Zhu <shixiong@databricks.com> Closes apache#15397 from zsxwing/SPARK-17834.

Fetch the earliest offsets manually in KafkaSource instead of countin…

95a0c96

…g on KafkaConsumer

zsxwing mentioned this pull request Oct 7, 2016

[SPARK-17782][STREAMING][KAFKA] eliminate race condition of poll twice #15387

Closed

clean up

9578555

koeninger reviewed Oct 13, 2016

View reviewed changes

Address

7986f18

asfgit closed this in 08eac35 Oct 13, 2016

zsxwing deleted the SPARK-17834 branch October 13, 2016 20:52

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-17834][SQL]Fetch the earliest offsets manually in KafkaSource instead of counting on KafkaConsumer #15397

[SPARK-17834][SQL]Fetch the earliest offsets manually in KafkaSource instead of counting on KafkaConsumer #15397

zsxwing commented Oct 7, 2016

zsxwing commented Oct 7, 2016

SparkQA commented Oct 8, 2016

koeninger commented Oct 8, 2016

zsxwing commented Oct 10, 2016

koeninger commented Oct 10, 2016

zsxwing commented Oct 12, 2016

SparkQA commented Oct 12, 2016

koeninger commented Oct 12, 2016

zsxwing commented Oct 13, 2016

koeninger left a comment

koeninger Oct 13, 2016

koeninger Oct 13, 2016

SparkQA commented Oct 13, 2016

koeninger commented Oct 13, 2016

zsxwing commented Oct 13, 2016

[SPARK-17834][SQL]Fetch the earliest offsets manually in KafkaSource instead of counting on KafkaConsumer #15397

[SPARK-17834][SQL]Fetch the earliest offsets manually in KafkaSource instead of counting on KafkaConsumer #15397

Conversation

zsxwing commented Oct 7, 2016

What changes were proposed in this pull request?

How was this patch tested?

zsxwing commented Oct 7, 2016

SparkQA commented Oct 8, 2016

koeninger commented Oct 8, 2016

zsxwing commented Oct 10, 2016

koeninger commented Oct 10, 2016

zsxwing commented Oct 12, 2016

SparkQA commented Oct 12, 2016

koeninger commented Oct 12, 2016

zsxwing commented Oct 13, 2016

koeninger left a comment

Choose a reason for hiding this comment

koeninger Oct 13, 2016

Choose a reason for hiding this comment

koeninger Oct 13, 2016

Choose a reason for hiding this comment

SparkQA commented Oct 13, 2016

koeninger commented Oct 13, 2016

zsxwing commented Oct 13, 2016