[SPARK-18020][Streaming][Kinesis] Checkpoint SHARD_END to finish reading closed shards #16213

maropu · 2016-12-08T11:15:02Z

What changes were proposed in this pull request?

This pr is to fix an issue occurred when resharding Kinesis streams; the resharding makes the KCL throw an exception because Spark does not checkpoint SHARD_END when finishing reading closed shards in KinesisRecordProcessor#shutdown. This bug finally leads to stopping subscribing new split (or merged) shards.

How was this patch tested?

Added a test in KinesisStreamSuite to check if it works well when splitting/merging shards.

…ction, this is workaround

SparkQA · 2016-12-08T11:47:49Z

Test build #69859 has finished for PR 16213 at commit 7c51da8.

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

maropu · 2016-12-08T12:28:03Z

I'm looking into the failure.

SparkQA · 2016-12-08T13:29:59Z

Test build #69860 has finished for PR 16213 at commit a5739b3.

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

SparkQA · 2016-12-08T18:37:41Z

Test build #69863 has finished for PR 16213 at commit e8a4e1d.

This patch fails from timeout after a configured wait of `250m`.
This patch merges cleanly.
This patch adds no public classes.

maropu · 2016-12-09T02:16:09Z

Jenkins, retest this please.

SparkQA · 2016-12-09T03:12:30Z

Test build #69902 has finished for PR 16213 at commit e8a4e1d.

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

brkyvz · 2016-12-09T19:28:36Z

.../src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/CheckpointerShim.java

+ */
+public class CheckpointerShim {
+
+  public static void shutdown(IRecordProcessorCheckpointer checkpointer)


Have you tried this call? https://github.com/awslabs/amazon-kinesis-client/blob/master/src/main/java/com/amazonaws/services/kinesis/clientlibrary/lib/worker/RecordProcessorCheckpointer.java#L129

It seems you may add a finalize method to KinesisCheckpointer where you call the IRecordProcessorCheckpointer.checkpoint(ExtendedSequenceNumber.SHARD_END.sequenceNumber, ExtendedSequenceNumber.SHARD_END.subSequenceNumber)

Yea, I tried by this version, but I got the same exception;

16/12/10 05:49:00 ERROR ShutdownTask: Application exception. java.lang.IllegalArgumentException: Sequence number must be numeric, but was SHARD_END at com.amazonaws.services.kinesis.clientlibrary.lib.worker.SequenceNumberValidator.validateSequenceNumber(SequenceNumberValidator.java:75) at com.amazonaws.services.kinesis.clientlibrary.lib.worker.RecordProcessorCheckpointer.checkpoint(RecordProcessorCheckpointer.java:120) at org.apache.spark.streaming.kinesis.KinesisCheckpointer.removeCheckpointer(KinesisCheckpointer.scala:77) at org.apache.spark.streaming.kinesis.KinesisReceiver.removeCheckpointer(KinesisReceiver.scala:258)

IIUC this is a kind of aws library bugs. I asked in the kinesis forum here, but I didn't get an answer.

brkyvz · 2016-12-09T19:33:56Z

external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala

+          "\nData received does not match data sent after splitting a shard")
+      }
+
+      val (shardToMerge, adjShared) = splitOpenShards match { case Seq(e1, e2) => (e1, e2) }


cleaner to do
val Seq(shardToMerge, adjShared) = splitOpenShards

probably you also meant adjShard. I would use adjacentShard. For a moment I was thinking adjective, shared adjective, what does that mean :P

Yea, I tried to use adjacentShard, but it returned null. I didn't why it behaved like this though, do you know that?

brkyvz · 2016-12-09T19:37:02Z

Left a couple comments. I don't like the idea of using KCL internal methods. Hopefully we won't need to.

maropu · 2016-12-10T06:02:57Z

Thanks for these comments! ya, I do not like this approach, too. But, since those who reshard streams always hit this issue and resharding is important for load-balancing in Kinesis streams (recently, a new API UpdteShardCount has been implemented for resharding), I'd like to fix this as soon as possible, but I couldn't find better approaches than this.

SparkQA · 2016-12-10T06:51:20Z

Test build #69956 has finished for PR 16213 at commit adf4dd6.

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

SparkQA · 2017-01-12T13:18:05Z

Test build #71260 has finished for PR 16213 at commit 1b89a0a.

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

SparkQA · 2017-01-12T14:08:18Z

Test build #71263 has finished for PR 16213 at commit 7f8336c.

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

SparkQA · 2017-01-12T15:23:19Z

Test build #71268 has finished for PR 16213 at commit 5f0e632.

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

maropu · 2017-01-12T15:27:40Z

@brkyvz I asked AWS guys and I found correct handling in this case (See: https://forums.aws.amazon.com/thread.jspa?threadID=244218). I fixed in this way and could you check this again? Thanks!

maropu · 2017-01-15T01:19:07Z

@brkyvz ping

brkyvz · 2017-01-18T00:43:35Z

Sorry, taking a look

maropu · 2017-01-18T00:54:22Z

Many thanks!

brkyvz · 2017-01-18T01:18:19Z

external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala

+      }
+      ssc.start()
+
+      val (testData1, testData2, testData3) = (1 to 10, 11 to 20, 21 to 30)


nit: instead of doing a tuple expansion, mind making these new lines

brkyvz · 2017-01-18T01:19:25Z

external/kinesis-asl/src/test/scala/org/apache/spark/streaming/kinesis/KinesisStreamSuite.scala

+          "\nData received does not match data sent after splitting a shard")
+      }
+
+      val Seq(shardToMerge, adjShared) = splitOpenShards


nit: adjShared -> adjShard?

brkyvz · 2017-01-18T01:20:11Z

This LGTM now! Thank you so much for checking with AWS and posting thread link as a comment.
cc @tdas to take a final look and possibly merge

SparkQA · 2017-01-18T03:20:08Z

Test build #71555 has finished for PR 16213 at commit b7ef9a9.

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

maropu · 2017-01-19T04:27:36Z

@tdas ping

maropu · 2017-01-21T01:51:57Z

@tdas ping

maropu · 2017-01-24T01:31:27Z

@brkyvz @tdas ping

Gauravshah · 2017-01-25T17:40:06Z

facing same issue randomly on prod, can I help in some way to push it ?

brkyvz · 2017-01-25T18:07:22Z

...rnal/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisCheckpointer.scala

+      // We must call `checkpoint()` with no parameter to finish reading shards.
+      // See an URL below for details:
+      // https://forums.aws.amazon.com/thread.jspa?threadID=244218
+      KinesisRecordProcessor.retryRandom(checkpointer.checkpoint(), 4, 100)


Can you run this if checkpointer != null. The docs mention that the checkpointer may be null.

Does the latest fix satisfy your intention?

brkyvz · 2017-01-25T18:09:58Z

@maropu One last comment. Then I'll merge this!

maropu · 2017-01-26T00:30:37Z

okay, I'll update soon!

brkyvz · 2017-01-26T00:50:41Z

LGTM! Pending tests.

maropu · 2017-01-26T01:24:11Z

Many thanks! Also, congrats, commiter

SparkQA · 2017-01-26T01:32:05Z

Test build #72007 has finished for PR 16213 at commit 17a5c3a.

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

brkyvz · 2017-01-26T01:36:41Z

Thanks a lot! Merging to master! (May take a while, going to be my first merge!)

maropu · 2017-01-26T01:37:44Z

great, haha! Many thanks!

…ing closed shards ## What changes were proposed in this pull request? This pr is to fix an issue occurred when resharding Kinesis streams; the resharding makes the KCL throw an exception because Spark does not checkpoint `SHARD_END` when finishing reading closed shards in `KinesisRecordProcessor#shutdown`. This bug finally leads to stopping subscribing new split (or merged) shards. ## How was this patch tested? Added a test in `KinesisStreamSuite` to check if it works well when splitting/merging shards. Author: Takeshi YAMAMURO <linguin.m.s@gmail.com> Closes apache#16213 from maropu/SPARK-18020.

Force advancing a position to SHARD_END by using the KCL internal fun…

e556d3c

…ction, this is workaround

maropu force-pushed the SPARK-18020 branch from 7c51da8 to a5739b3 Compare December 8, 2016 12:48

Add tests for splitting and merging shards

e8a4e1d

maropu force-pushed the SPARK-18020 branch from a5739b3 to e8a4e1d Compare December 8, 2016 14:24

maropu mentioned this pull request Dec 9, 2016

[SPARK-18620][Streaming][Kinesis] Flatten input rates in timeline for streaming + kinesis #16114

Closed

brkyvz reviewed Dec 9, 2016

View reviewed changes

Apply comments

adf4dd6

Apply comments

1b89a0a

Fix test failures

5f0e632

maropu force-pushed the SPARK-18020 branch from 7f8336c to 5f0e632 Compare January 12, 2017 14:23

brkyvz reviewed Jan 18, 2017

View reviewed changes

Apply comments

b7ef9a9

brkyvz reviewed Jan 25, 2017

View reviewed changes

Guard null pointer exceptions

17a5c3a

asfgit closed this in 256a3a8 Jan 26, 2017

[SPARK-18020][Streaming][Kinesis] Checkpoint SHARD_END to finish reading closed shards #16213

[SPARK-18020][Streaming][Kinesis] Checkpoint SHARD_END to finish reading closed shards #16213

Conversation

maropu commented Dec 8, 2016

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Dec 8, 2016

maropu commented Dec 8, 2016

SparkQA commented Dec 8, 2016

SparkQA commented Dec 8, 2016

maropu commented Dec 9, 2016

SparkQA commented Dec 9, 2016

Choose a reason for hiding this comment

maropu Dec 10, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brkyvz Dec 9, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brkyvz commented Dec 9, 2016

maropu commented Dec 10, 2016

SparkQA commented Dec 10, 2016

SparkQA commented Jan 12, 2017

SparkQA commented Jan 12, 2017

SparkQA commented Jan 12, 2017

maropu commented Jan 12, 2017

maropu commented Jan 15, 2017

brkyvz commented Jan 18, 2017

maropu commented Jan 18, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brkyvz commented Jan 18, 2017

SparkQA commented Jan 18, 2017

maropu commented Jan 19, 2017

maropu commented Jan 21, 2017

maropu commented Jan 24, 2017

Gauravshah commented Jan 25, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brkyvz commented Jan 25, 2017

maropu commented Jan 26, 2017

brkyvz commented Jan 26, 2017

maropu commented Jan 26, 2017

SparkQA commented Jan 26, 2017

brkyvz commented Jan 26, 2017

maropu commented Jan 26, 2017

maropu Dec 10, 2016 •

edited

Loading

brkyvz Dec 9, 2016 •

edited

Loading