SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver #10203

jbonofre · 2015-12-08T16:52:45Z

No description provided.

EugenCepoi · 2015-12-08T19:13:26Z

Why don't you just replace the use of SynchronizedMap in KinesisReceiver with a ConcurrentHashMap instead?

jbonofre · 2015-12-08T19:20:27Z

There are two things:

as you said, we could use a Java ConcurrentHashMap in KinesisReceiver (changing couple of code)
but it still means that users can't use Kryo serializer with SynchronizedMap trait (generally speaking). That's why I proposed a more "generic" solution.

Thought ?

EugenCepoi · 2015-12-08T19:31:46Z

That problem actually applies to all types for which Kryo provides a default ser/de. Mostly because kryo will try to deserialize to the type known during registration and this syntax new Foo with Bar generates a class. I am wondering if this can't also occur for plain pojo/poso for which kryo will use it generic deser system. In short I think it is better to just use the java impl and eventually report that kind of problem to kryo chil project. Anyway SynchronizedMap is deperecated in scala 2.11.

jbonofre · 2015-12-08T19:36:46Z

Good point. Let me upgrade KinesisReceiver to use Java ConcurrentHashMap implementation in this PR. We will see what the others think about this.

srowen · 2015-12-09T14:54:22Z

Yes, the simpler solution is better here. Use ConcurrentHashMap.

jbonofre · 2015-12-09T15:47:34Z

All right. I'm updating the PR.

Maybe it could make sense to inform people who use Kryo that the SynchronizedMap trait is "lost".

WDYT ?

jbonofre · 2015-12-09T17:35:54Z

PR rebased and updated to use Java ConcurrentHashMap.

I removed the change on the KryoSeralizer to deal with SynchronizedMap trait.

srowen · 2015-12-09T17:43:07Z

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala

@@ -222,7 +222,7 @@ private[kinesis] class KinesisReceiver[T](

  /** Get the latest sequence number for the given shard that can be checkpointed through KCL */
  private[kinesis] def getLatestSeqNumToCheckpoint(shardId: String): Option[String] = {
-    shardIdToLatestStoredSeqNum.get(shardId)
+    return Option[String]{ shardIdToLatestStoredSeqNum.get(shardId) }


This can be simplified to just Option(shardIdToLatestStoredSeqNum.get(shardId))

Good point Sean. Let me improve this !

Thanks !

srowen · 2015-12-09T17:43:33Z

Aside from the two instances of that comment, looks OK

EugenCepoi · 2015-12-09T18:06:30Z

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala

@@ -222,7 +222,7 @@ private[kinesis] class KinesisReceiver[T](

  /** Get the latest sequence number for the given shard that can be checkpointed through KCL */
  private[kinesis] def getLatestSeqNumToCheckpoint(shardId: String): Option[String] = {
-    shardIdToLatestStoredSeqNum.get(shardId)
+    return Option(shardIdToLatestStoredSeqNum.get(shardId))


You can drop the return keyword

Good catch, I forgot this cleanup. Thanks !

EugenCepoi · 2015-12-09T18:14:07Z

Looks good

srowen · 2015-12-09T18:24:28Z

extras/kinesis-asl/src/main/scala/org/apache/spark/streaming/kinesis/KinesisReceiver.scala

@@ -124,8 +125,7 @@ private[kinesis] class KinesisReceiver[T](
  private val seqNumRangesInCurrentBlock = new mutable.ArrayBuffer[SequenceNumberRange]

  /** Sequence number ranges of data added to each generated block */
-  private val blockIdToSeqNumRanges = new mutable.HashMap[StreamBlockId, SequenceNumberRanges]
-    with mutable.SynchronizedMap[StreamBlockId, SequenceNumberRanges]
+  private val blockIdToSeqNumRanges = new ConcurrentHashMap[StreamBlockId, SequenceNumberRanges]


I don't think you need to fix this -- but I think the style convention is to use () when the invocation has a side effect and I'd argue that constructors always do. I should have said it earlier but don't know that it's worth changing as the original call didn't either.

Oh, thanks for this reminder Sean. You are right, I used the same syntax as in the original code. Let me know if you want I change this.

It's probably always a good idea to use ConcurrentHashMap instead of the mixed-in trait. The typesafe people themselves deprecated the trait and said it's unreliable and recommended that users use java's map instead.

SparkQA · 2015-12-09T19:06:34Z

Test build #2190 has finished for PR 10203 at commit d80cf35.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):\n * public class JavaIndexToStringExample\n

srowen · 2015-12-10T12:57:16Z

@jbonofre can you update the title to reflect the change? possibly description too

SparkQA · 2015-12-10T14:47:59Z

Test build #2194 has started for PR 10203 at commit d80cf35.

jbonofre · 2015-12-10T15:40:51Z

Sure. You mean the PR title or also the commit comment ?

srowen · 2015-12-10T15:56:23Z

PR title/description. The squashed commit gets a new message and the squashed commit descriptions look OK anyway.

jbonofre · 2015-12-10T15:58:54Z

All right, let me do it.

andrewor14 · 2015-12-11T19:46:46Z

retest this please. The changes here LGTM

jbonofre · 2015-12-11T21:37:02Z

Thanks @andrewor14 (again ;)). Let me retest it.

…dMap trait in order to avoid Kryo serializer issue

… function

jbonofre · 2015-12-11T21:50:58Z

Tests OK on my box. The Jenkins test failure doesn't look related.

SparkQA · 2015-12-11T22:55:53Z

Test build #2208 has finished for PR 10203 at commit 5cec007.

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

srowen · 2015-12-12T08:52:25Z

Merged to master/1.6

…rait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver Author: Jean-Baptiste Onofré <jbonofre@apache.org> Closes #10203 from jbonofre/SPARK-11193. (cherry picked from commit 03138b6) Signed-off-by: Sean Owen <sowen@cloudera.com>

jbonofre force-pushed the SPARK-11193 branch 2 times, most recently from 313517f to caa4363 Compare December 9, 2015 17:35

srowen reviewed Dec 9, 2015
View reviewed changes

EugenCepoi reviewed Dec 9, 2015
View reviewed changes

srowen reviewed Dec 9, 2015
View reviewed changes

jbonofre changed the title ~~SPARK-11193 - Hack to support SynchronizedMap trait in Kryo serializer~~ SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver Dec 10, 2015

jbonofre force-pushed the SPARK-11193 branch 2 times, most recently from 9cad42d to 67aa4e6 Compare December 11, 2015 16:44

jbonofre added 3 commits December 11, 2015 22:50

SPARK-11193 - Use a Java ConcurrentHashMap instead of the Synchronize…

403d874

…dMap trait in order to avoid Kryo serializer issue

SPARK-11193 - Improve Option usage

61221af

SPARK-11193 - Cleanup return keyword in getLatestSeqNumToCheckpoint()…

5cec007

… function

jbonofre force-pushed the SPARK-11193 branch from 67aa4e6 to 5cec007 Compare December 11, 2015 21:50

asfgit closed this in 03138b6 Dec 12, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver #10203

SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver #10203

jbonofre commented Dec 8, 2015

EugenCepoi commented Dec 8, 2015

jbonofre commented Dec 8, 2015

EugenCepoi commented Dec 8, 2015

jbonofre commented Dec 8, 2015

srowen commented Dec 9, 2015

jbonofre commented Dec 9, 2015

jbonofre commented Dec 9, 2015

srowen Dec 9, 2015

jbonofre Dec 9, 2015

srowen commented Dec 9, 2015

EugenCepoi Dec 9, 2015

jbonofre Dec 9, 2015

EugenCepoi commented Dec 9, 2015

srowen Dec 9, 2015

jbonofre Dec 9, 2015

andrewor14 Dec 11, 2015

SparkQA commented Dec 9, 2015

srowen commented Dec 10, 2015

SparkQA commented Dec 10, 2015

jbonofre commented Dec 10, 2015

srowen commented Dec 10, 2015

jbonofre commented Dec 10, 2015

andrewor14 commented Dec 11, 2015

jbonofre commented Dec 11, 2015

jbonofre commented Dec 11, 2015

SparkQA commented Dec 11, 2015

srowen commented Dec 12, 2015

SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver #10203

SPARK-11193 - Use Java ConcurrentHashMap instead of SynchronizedMap trait in order to avoid ClassCastException due to KryoSerializer in KinesisReceiver #10203

Conversation

jbonofre commented Dec 8, 2015

EugenCepoi commented Dec 8, 2015

jbonofre commented Dec 8, 2015

EugenCepoi commented Dec 8, 2015

jbonofre commented Dec 8, 2015

srowen commented Dec 9, 2015

jbonofre commented Dec 9, 2015

jbonofre commented Dec 9, 2015

srowen Dec 9, 2015

Choose a reason for hiding this comment

jbonofre Dec 9, 2015

Choose a reason for hiding this comment

srowen commented Dec 9, 2015

EugenCepoi Dec 9, 2015

Choose a reason for hiding this comment

jbonofre Dec 9, 2015

Choose a reason for hiding this comment

EugenCepoi commented Dec 9, 2015

srowen Dec 9, 2015

Choose a reason for hiding this comment

jbonofre Dec 9, 2015

Choose a reason for hiding this comment

andrewor14 Dec 11, 2015

Choose a reason for hiding this comment

SparkQA commented Dec 9, 2015

srowen commented Dec 10, 2015

SparkQA commented Dec 10, 2015

jbonofre commented Dec 10, 2015

srowen commented Dec 10, 2015

jbonofre commented Dec 10, 2015

andrewor14 commented Dec 11, 2015

jbonofre commented Dec 11, 2015

jbonofre commented Dec 11, 2015

SparkQA commented Dec 11, 2015

srowen commented Dec 12, 2015