[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are SpecificOffsets #17209

uncleGen · 2017-03-08T12:05:14Z

What changes were proposed in this pull request?

When using the KafkaSource with Structured Streaming, consumer assignments are not what the user expects if startingOffsets is set to an explicit set of topics/partitions in JSON where the topic(s) happen to have uppercase characters. When StartingOffsets is constructed, the original string value from options is transformed toLowerCase to make matching on "earliest" and "latest" case insensitive. However, the toLowerCase JSON is passed to SpecificOffsets for the terminal condition, so topic names may not be what the user intended by the time assignments are made with the underlying KafkaConsumer.

KafkaSourceProvider.scala:

val startingOffsets = caseInsensitiveParams.get(STARTING_OFFSETS_OPTION_KEY).map(_.trim.toLowerCase) match {
    case Some("latest") => LatestOffsets
    case Some("earliest") => EarliestOffsets
    case Some(json) => SpecificOffsets(JsonUtils.partitionOffsets(json))
    case None => LatestOffsets
  }

Thank @cbowden for reporting.

How was this patch tested?

Jenkins

SparkQA · 2017-03-08T12:29:12Z

Test build #74205 has finished for PR 17209 at commit e2a26bf.

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

cbcwebdev

lgtm, perhaps as a slightly additional enhancement the underlying tests could be modified to include some uppercase characters in their topic name(s) for at least one case

uncleGen · 2017-03-09T03:49:26Z

cc @zsxwing

SparkQA · 2017-03-09T04:07:43Z

Test build #74237 has finished for PR 17209 at commit 9402f9d.

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

zsxwing · 2017-03-10T01:37:05Z

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

-        case Some("latest") => LatestOffsetRangeLimit
-        case Some("earliest") => EarliestOffsetRangeLimit
+      caseInsensitiveParams.get(STARTING_OFFSETS_OPTION_KEY).map(_.trim) match {
+        case a @ Some(offset) if offset.toLowerCase.equals("latest") => LatestOffsetRangeLimit


case a @ Some(offset) if offset.toLowerCase.equals("latest") -> case Some(offset) if offset.toLowerCase == "latest".

Please also fix other places.

Could you just write a utility method to get offsets from caseInsensitiveParams? Then you can just test this method.

SparkQA · 2017-03-10T02:37:21Z

Test build #74291 has finished for PR 17209 at commit a37d11d.

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

zsxwing · 2017-03-10T23:02:00Z

Could you add a utility method like this?

  def getKafkaOffsetRangeLimit(
      params: Map[String, String],
      key: String,
      default: KafkaOffsetRangeLimit) : KafkaOffsetRangeLimit = {
    params.get(key).map(_.trim) match {
      case Some(offset) if offset.toLowerCase == "latest" => LatestOffsetRangeLimit
      case Some(offset) if offset.toLowerCase == "earliest" => EarliestOffsetRangeLimit
      case Some(json) => SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json))
      case None => default
    }
  }

Then you can reuse the method and only test it, e.g.,

    val startingOffsets = getKafkaOffsetRangeLimit(
      caseInsensitiveParams,
      STARTING_OFFSETS_OPTION_KEY,
      EarliestOffsetRangeLimit
    )

    // Batch specific options
    startingOffsets match {
      case EarliestOffsetRangeLimit => // good to go
      case LatestOffsetRangeLimit =>
        throw new IllegalArgumentException("starting offset can't be latest " +
          "for batch queries on Kafka")
      case SpecificOffsetRangeLimit(partitionOffsets) =>
        partitionOffsets.foreach {
          case (tp, off) if off == KafkaOffsetRangeLimit.LATEST =>
            throw new IllegalArgumentException(s"startingOffsets for $tp can't " +
              "be latest for batch queries on Kafka")
          case _ => // ignore
        }
    }

uncleGen · 2017-03-11T02:52:45Z

@zsxwing done，but forgot to push，I will update it as soon as possible when I connect to internet.

SparkQA · 2017-03-11T07:10:44Z

Test build #74380 has finished for PR 17209 at commit 50ef0e1.

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

uncleGen · 2017-03-11T07:18:12Z

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

-  private val STARTING_OFFSETS_OPTION_KEY = "startingoffsets"
-  private val ENDING_OFFSETS_OPTION_KEY = "endingoffsets"
+  private[kafka010] val STARTING_OFFSETS_OPTION_KEY = "startingoffsets"
+  private[kafka010] val ENDING_OFFSETS_OPTION_KEY = "endingoffsets"
  private val FAIL_ON_DATA_LOSS_OPTION_KEY = "failondataloss"


change for unit test

zsxwing

Looks pretty good. Just some minor issues.

zsxwing · 2017-03-12T04:45:42Z

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

-        case None => EarliestOffsetRangeLimit
+    val startingRelationOffsets = KafkaSourceProvider.getKafkaOffsetRangeLimit(
+      caseInsensitiveParams, STARTING_OFFSETS_OPTION_KEY, EarliestOffsetRangeLimit) match {
+        case earliest @ EarliestOffsetRangeLimit => earliest


startingRelationOffsets won't be latest since it's checked in validateBatchOptions.
Why not just:

val startingRelationOffsets = KafkaSourceProvider.getKafkaOffsetRangeLimit( caseInsensitiveParams, STARTING_OFFSETS_OPTION_KEY, EarliestOffsetRangeLimit) assert(startingRelationOffsets != LatestOffsetRangeLimit)

👍 much more simple

zsxwing · 2017-03-12T04:45:50Z

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

-        case None => LatestOffsetRangeLimit
+    val endingRelationOffsets = KafkaSourceProvider.getKafkaOffsetRangeLimit(caseInsensitiveParams,
+      ENDING_OFFSETS_OPTION_KEY, LatestOffsetRangeLimit) match {
+        case latest @ LatestOffsetRangeLimit => latest


Same as above

zsxwing · 2017-03-12T04:46:12Z

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

        throw new IllegalArgumentException("starting offset can't be latest " +
          "for batch queries on Kafka")
-      case Some(json) => (SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json)))
-        .partitionOffsets.foreach {
+      case specific: SpecificOffsetRangeLimit =>


nit: case SpecificOffsetRangeLimit(partitionOffsets) =>

zsxwing · 2017-03-12T04:46:18Z

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

-      case Some(json) => (SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json)))
-        .partitionOffsets.foreach {
+      case LatestOffsetRangeLimit => // good to go
+      case specific: SpecificOffsetRangeLimit =>


nit: case SpecificOffsetRangeLimit(partitionOffsets) =>

zsxwing · 2017-03-12T04:51:12Z

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala

@@ -606,6 +607,36 @@ class KafkaSourceSuite extends KafkaSourceTest {
    assert(query.exception.isEmpty)
  }

+  for((optionKey, optionValue, answer) <- Seq(


nit: move the for loop into the test. Not need to create many tests here.

zsxwing · 2017-03-12T04:51:22Z

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala

+    }
+  }
+
+  for((optionKey, answer) <- Seq(


Same as above

zsxwing · 2017-03-12T04:51:28Z

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala

+        answer
+      )
+
+      assert(offset == answer)


nit == => ===

SparkQA · 2017-03-12T10:16:10Z

Test build #74399 has finished for PR 17209 at commit be9f87e.

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

…e SpecificOffsets When using the KafkaSource with Structured Streaming, consumer assignments are not what the user expects if startingOffsets is set to an explicit set of topics/partitions in JSON where the topic(s) happen to have uppercase characters. When StartingOffsets is constructed, the original string value from options is transformed toLowerCase to make matching on "earliest" and "latest" case insensitive. However, the toLowerCase JSON is passed to SpecificOffsets for the terminal condition, so topic names may not be what the user intended by the time assignments are made with the underlying KafkaConsumer. KafkaSourceProvider.scala: ``` val startingOffsets = caseInsensitiveParams.get(STARTING_OFFSETS_OPTION_KEY).map(_.trim.toLowerCase) match { case Some("latest") => LatestOffsets case Some("earliest") => EarliestOffsets case Some(json) => SpecificOffsets(JsonUtils.partitionOffsets(json)) case None => LatestOffsets } ``` Thank cbowden for reporting. Jenkins Author: uncleGen <hustyugm@gmail.com> Closes #17209 from uncleGen/SPARK-19853. (cherry picked from commit 0a4d06a) Signed-off-by: Shixiong Zhu <shixiong@databricks.com>

zsxwing · 2017-03-13T00:55:41Z

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala

@@ -606,6 +607,24 @@ class KafkaSourceSuite extends KafkaSourceTest {
    assert(query.exception.isEmpty)
  }

+  test("test to get offsets from case insensitive parameters") {


nit: remove test to. I will fix it when merging your PR.

zsxwing · 2017-03-13T00:55:58Z

Thanks. Merging to master and 2.1.

Uppercase Kafka topics fail when startingOffsets are SpecificOffsets

e2a26bf

cbcwebdev approved these changes Mar 8, 2017

View reviewed changes

add unit test

9402f9d

cbcwebdev approved these changes Mar 9, 2017

View reviewed changes

zsxwing reviewed Mar 10, 2017

View reviewed changes

address @zsxwing's comments

a37d11d

address @zsxwing's comments

50ef0e1

uncleGen commented Mar 11, 2017

View reviewed changes

zsxwing requested changes Mar 12, 2017

View reviewed changes

uncleGen added 2 commits March 12, 2017 17:03

Merge branch 'master' into SPARK-19853

8ded62a

address comments from @zsxwing

be9f87e

zsxwing reviewed Mar 13, 2017

View reviewed changes

asfgit closed this in 0a4d06a Mar 13, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are SpecificOffsets #17209

[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are SpecificOffsets #17209

uncleGen commented Mar 8, 2017 •

edited

Loading

SparkQA commented Mar 8, 2017

cbcwebdev left a comment •

edited

Loading

uncleGen commented Mar 9, 2017

SparkQA commented Mar 9, 2017

zsxwing Mar 10, 2017

zsxwing Mar 10, 2017 •

edited

Loading

SparkQA commented Mar 10, 2017

zsxwing commented Mar 10, 2017

uncleGen commented Mar 11, 2017

SparkQA commented Mar 11, 2017

uncleGen Mar 11, 2017

zsxwing left a comment

zsxwing Mar 12, 2017

uncleGen Mar 12, 2017

zsxwing Mar 12, 2017

zsxwing Mar 12, 2017

zsxwing Mar 12, 2017

zsxwing Mar 12, 2017

zsxwing Mar 12, 2017

zsxwing Mar 12, 2017

SparkQA commented Mar 12, 2017

zsxwing Mar 13, 2017

uncleGen Mar 13, 2017

zsxwing commented Mar 13, 2017

[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are SpecificOffsets #17209

[SPARK-19853][SS] uppercase kafka topics fail when startingOffsets are SpecificOffsets #17209

Conversation

uncleGen commented Mar 8, 2017 • edited Loading

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Mar 8, 2017

cbcwebdev left a comment • edited Loading

Choose a reason for hiding this comment

uncleGen commented Mar 9, 2017

SparkQA commented Mar 9, 2017

Choose a reason for hiding this comment

zsxwing Mar 10, 2017 • edited Loading

Choose a reason for hiding this comment

SparkQA commented Mar 10, 2017

zsxwing commented Mar 10, 2017

uncleGen commented Mar 11, 2017

SparkQA commented Mar 11, 2017

Choose a reason for hiding this comment

zsxwing left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Mar 12, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zsxwing commented Mar 13, 2017

uncleGen commented Mar 8, 2017 •

edited

Loading

cbcwebdev left a comment •

edited

Loading

zsxwing Mar 10, 2017 •

edited

Loading