[SPARK-23539][SS] Add support for Kafka headers in Structured Streaming #22282

dongjinleekr · 2018-08-30T12:28:20Z

What changes were proposed in this pull request?

This update adds support for Kafka Headers functionality in Structured Streaming.

How was this patch tested?

With following unit tests:

KafkaRelationSuite: "default starting and ending offsets with headers" (new)
KafkaSinkSuite: "batch - write to kafka" (updated)

dongjinleekr · 2018-08-30T12:36:16Z

As you can see, this PR consists of 3 parts:

Extend UnsafeArrayData, UnsafeMapData (commit 1~6)
Implement Kafka Headers functionality (commit 7, 10)
Update unit tests (commit 8, 9)

I have the following questions:

Should I separate group 1 as a separated issue?
I found that KafkaSourceSuiteBase's 'Kafka column types' test is missing a select expression. The weird thing is that it works before the update but does not work after the update. (It is why the last commit was added - without specification, this test does not pass.) Is this intended one? Or, Do I misunderstanding something?

Please have a look when you are free. Thanks.

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

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeArrayData.java

HeartSaVioR

The change mostly looks good. Left some comments to point some possible minors as well as style nits.

...-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala

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

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

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

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

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

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

## What changes were proposed in this pull request? Fix unused imports & outdated comments on `kafka-0-10-sql` module. (Found while I was working on [SPARK-23539](#22282)) ## How was this patch tested? Existing unit tests. Closes #22342 from dongjinleekr/feature/fix-kafka-sql-trivials. Authored-by: Lee Dongjin <dongjin@apache.org> Signed-off-by: Sean Owen <sean.owen@databricks.com> (cherry picked from commit 458f501) Signed-off-by: Sean Owen <sean.owen@databricks.com>

## What changes were proposed in this pull request? Fix unused imports & outdated comments on `kafka-0-10-sql` module. (Found while I was working on [SPARK-23539](#22282)) ## How was this patch tested? Existing unit tests. Closes #22342 from dongjinleekr/feature/fix-kafka-sql-trivials. Authored-by: Lee Dongjin <dongjin@apache.org> Signed-off-by: Sean Owen <sean.owen@databricks.com>

dongjinleekr · 2018-09-10T23:45:38Z

retest this please.

SparkQA · 2018-09-11T03:48:48Z

Test build #95906 has finished for PR 22282 at commit 220bd0a.

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

dongjinleekr · 2018-11-12T13:06:55Z

cc/ @zsxwing @tdas @dongjoon-hyun @srowen Rebased onto the latest master. Please have a look when you are free. Thanks in advance.

SparkQA · 2018-11-12T16:40:31Z

Test build #98726 has finished for PR 22282 at commit 2c48aae.

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

srowen

Some minor feedback to start

...nal/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDataConsumerSuite.scala

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

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

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/UnsafeArraySuite.scala

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

...nal/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDataConsumerSuite.scala

SparkQA · 2019-02-10T08:05:02Z

Test build #102141 has finished for PR 22282 at commit 9fad0ca.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

dongjinleekr · 2019-02-11T04:23:01Z

retest this please.

@HyukjinKwon Could you have a look when you are free? cc/ @kiszk @zsxwing @tdas

HyukjinKwon · 2019-02-11T05:01:27Z

retest this please

SparkQA · 2019-02-11T06:00:07Z

Test build #102174 has finished for PR 22282 at commit 9fad0ca.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

zsxwing · 2019-02-11T18:10:22Z

If a user uses a Kafka cluster which runs using an old version that doesn't support Kafka headers, will their query fail?

HeartSaVioR · 2019-09-10T09:40:19Z

@zsxwing @srowen
Could you please take another round of review? The patch now looks to address all of review comments and rebased with master.

srowen

A few small style items

srowen · 2019-09-10T14:01:02Z

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

@@ -131,9 +145,26 @@ private[kafka010] abstract class KafkaRowWriter(
        throw new IllegalStateException(s"${KafkaWriter.VALUE_ATTRIBUTE_NAME} " +
          s"attribute unsupported type ${t.catalogString}")
    }
+    val headersExpression = inputSchema
+      .find(_.name == KafkaWriter.HEADERS_ATTRIBUTE_NAME).getOrElse(
+      Literal(CatalystTypeConverters.convertToCatalyst(null),


Should this be indented further?

I tried, but the formatter reverts the indention to the current status.

The style checker or something else?

The code formatter of the IDE. Also, it passes the style checker of mvn.

...nal/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDataConsumerSuite.scala

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

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

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

dongjinleekr · 2019-09-11T03:26:40Z

@srowen Fixed, with removing redundant new in KafkaDataConsumerSuite.

SparkQA · 2019-09-11T03:58:15Z

Test build #110456 has finished for PR 22282 at commit 243809f.

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

srowen · 2019-09-12T13:53:54Z

I think it's OK @dongjinleekr , just needs a rebase now.

…TRUCT<key:STRING,value:BINARY>>)'

Move projection methods in `KafkaOffsetReader` to `KafkaRecordToRowConverter`.

dongjinleekr · 2019-09-13T00:29:42Z

@srowen Here is the rebase. Thanks for reviewing! 😄

SparkQA · 2019-09-13T01:02:35Z

Test build #110550 has finished for PR 22282 at commit de02de4.

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

srowen · 2019-09-13T17:33:00Z

Merged to master

dongjoon-hyun · 2019-09-13T17:35:47Z

Great! @dongjinleekr , @srowen , @HeartSaVioR , @zsxwing !

HeartSaVioR · 2019-09-13T20:33:37Z

Finally! Well done, @dongjinleekr !

## What changes were proposed in this pull request? This update adds support for Kafka Headers functionality in Structured Streaming. ## How was this patch tested? With following unit tests: - KafkaRelationSuite: "default starting and ending offsets with headers" (new) - KafkaSinkSuite: "batch - write to kafka" (updated) Closes apache#22282 from dongjinleekr/feature/SPARK-23539. Lead-authored-by: Lee Dongjin <dongjin@apache.org> Co-authored-by: Jungtaek Lim <kabhwan@gmail.com> Signed-off-by: Sean Owen <sean.owen@databricks.com>

gaborgsomogyi · 2019-09-27T12:34:54Z

Just gone through, good job @dongjinleekr

This comment has been minimized.

Sign in to view

tedyu reviewed Aug 30, 2018

View reviewed changes

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaWriteTask.scala Show resolved Hide resolved

xuanyuanking reviewed Aug 30, 2018

View reviewed changes

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeArrayData.java Outdated Show resolved Hide resolved

dongjinleekr force-pushed the feature/SPARK-23539 branch from 229aac8 to 8a5ef14 Compare August 31, 2018 12:52

This comment has been minimized.

Sign in to view

HeartSaVioR reviewed Sep 3, 2018

View reviewed changes

dongjinleekr force-pushed the feature/SPARK-23539 branch from 2254009 to 253a894 Compare September 4, 2018 11:34

This comment has been minimized.

Sign in to view

dongjinleekr mentioned this pull request Sep 5, 2018

[MINOR][SS] Fix kafka-0-10-sql trivials #22342

Closed

dongjinleekr force-pushed the feature/SPARK-23539 branch from 253a894 to b0f64e9 Compare September 5, 2018 22:00

This comment has been minimized.

Sign in to view

dongjinleekr force-pushed the feature/SPARK-23539 branch from 220bd0a to 2c48aae Compare November 12, 2018 13:02

srowen requested changes Jan 13, 2019

View reviewed changes

dongjinleekr force-pushed the feature/SPARK-23539 branch from 2c48aae to 9fad0ca Compare February 10, 2019 07:06

srowen reviewed Sep 10, 2019

View reviewed changes

dongjinleekr and others added 8 commits September 13, 2019 09:28

[SPARK-23539][SS] Add support for Kafka headers in Structured Streaming

2935dae

Minor improvements

0ce84b3

Remove postfix notation in KafkaDataConsumerSuite

f580431

parameters -> caseInsensitiveParameters

2330679

Remove redundant casting for headers column: 'CAST(headers as ARRAY<S…

9949c4e

…TRUCT<key:STRING,value:BINARY>>)'

Applying review comment from zsxwing properly - critical path (#1)

1d926c4

Move projection methods in `KafkaOffsetReader` to `KafkaRecordToRowConverter`.

Fix rebasing conflicts

e458a60

Trivial style improvements

de02de4

dongjinleekr force-pushed the feature/SPARK-23539 branch from 243809f to de02de4 Compare September 13, 2019 00:29

srowen closed this in 1675d51 Sep 13, 2019

HeartSaVioR mentioned this pull request Sep 15, 2019

[SPARK-23539][SS][FOLLOWUP][TESTS] Add UT to ensure existing query doesn't break with default conf of includeHeaders #25792

Closed

HeartSaVioR mentioned this pull request Mar 3, 2022

[SPARK-38323][SQL][Streaming] Support the hidden file metadata in Streaming #35676

Closed

[SPARK-23539][SS] Add support for Kafka headers in Structured Streaming #22282

[SPARK-23539][SS] Add support for Kafka headers in Structured Streaming #22282

Uh oh!

Conversation

dongjinleekr commented Aug 30, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

dongjinleekr commented Aug 30, 2018

Uh oh!

This comment has been minimized.

Uh oh!

Uh oh!

This comment has been minimized.

HeartSaVioR left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

This comment has been minimized.

dongjinleekr commented Sep 10, 2018

Uh oh!

SparkQA commented Sep 11, 2018

Uh oh!

dongjinleekr commented Nov 12, 2018

Uh oh!

SparkQA commented Nov 12, 2018

Uh oh!

srowen left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Feb 10, 2019

Uh oh!

dongjinleekr commented Feb 11, 2019

Uh oh!

HyukjinKwon commented Feb 11, 2019

Uh oh!

SparkQA commented Feb 11, 2019

Uh oh!

zsxwing commented Feb 11, 2019

Uh oh!

HeartSaVioR commented Sep 10, 2019

Uh oh!

srowen left a comment

Choose a reason for hiding this comment

Uh oh!

srowen Sep 10, 2019

Choose a reason for hiding this comment

Uh oh!

dongjinleekr Sep 11, 2019

Choose a reason for hiding this comment

Uh oh!

srowen Sep 11, 2019

Choose a reason for hiding this comment

Uh oh!

dongjinleekr Sep 13, 2019

Choose a reason for hiding this comment

Uh oh!

Uh oh!