Should use ConcurrentHashMap instead of HashMap in ParquetMetadataConverter

[asfimport]

When using parquet in Spark Environment, sometimes got hang with following thread dump:
"Executor task launch worker-0" daemon prio=10 tid=0x000000004073d000 nid=0xd6c5 runnable [0x00007ff3fda40000]
java.lang.Thread.State: RUNNABLE
at java.util.HashMap.get(HashMap.java:303)
at parquet.format.converter.ParquetMetadataConverter.fromFormatEncodings(ParquetMetadataConverter.java:218)
at parquet.format.converter.ParquetMetadataConverter.fromParquetMetadata(ParquetMetadataConverter.java:543)
at parquet.format.converter.ParquetMetadataConverter.readParquetMetadata(ParquetMetadataConverter.java:520)
at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:426)
at parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:381)
at parquet.hadoop.ParquetRecordReader.initializeInternalReader(ParquetRecordReader.java:161)
at parquet.hadoop.ParquetRecordReader.initialize(ParquetRecordReader.java:138)
at org.apache.spark.rdd.NewHadoopRDD$$anon$1.(NewHadoopRDD.scala:135)
at org.apache.spark.rdd.NewHadoopRDD.compute(NewHadoopRDD.scala:107)
at org.apache.spark.rdd.NewHadoopRDD.compute(NewHadoopRDD.scala:69)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.rdd.FlatMappedRDD.compute(FlatMappedRDD.scala:33)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:35)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:230)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
at org.apache.spark.scheduler.Task.run(Task.scala:56)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:662)

From the source code of ParquetMetadataConverter:

private Map> encodingLists = new HashMap>();

It use HashMap instead of ConcurrentHashMap. Because HashMap is not thread safe and can cause hang when run in multithread environment. So it need to change to ConcurrentHashMap

Environment: Spark 1.2.1, CentOS 6.4
Reporter: Tony Yan
Assignee: Alex Levenson / @isnotinvain

Related issues:

• Release Parquet 1.8.0 (blocks)

PRs and other links:

• PR-220

_{Note: This issue was originally created as PARQUET-284. Please see the migration documentation for further details.}

[asfimport]

Ryan Blue / @rdblue:
Looks like we should probably change that HashMap and make it static. The reason why it wasn't concurrent is that it is an instance field. But ParquetFileReader keeps a static ParquetMetadataConverter around, assuming that it is thread-safe. At the same time, we should make sure the rest of the class is thread-safe.

[asfimport]

Alex Levenson / @isnotinvain:
How important is it to cache this state? Simply not cacheing this (and any other instance variables) will make this class thread safe / stateless.

Or, how about using a ThreadLocal cache? Does recomputing this per-thread ruin the performance gain?

[asfimport]

Alex Levenson / @isnotinvain:
Update: the cache was actually never being used as a cache... so I've removed the cacheing entirely, and made this class stateless + static

[asfimport]

Alex Levenson / @isnotinvain:
[~yanzt], I've sent a PR to clean up this class and remove it's use of a HashMap.

However, I don't think this was your root issue – do you know how concurrent access to a HashMap could deadlock? HashMap is not synchronized, so concurrent access probably doesn't involve locking (for dead lock) and would probably result in undefined return values / behavior or concurrent modification exceptions.

Or is there some way it can get into an infinite loop that you know of?

[asfimport]

Alex Levenson / @isnotinvain:
welp, I'm wrong: https://gist.github.com/rednaxelafx/1081908
It can cause an infinite loop!

[asfimport]

Tony Yan:
Thanks for the fix.

[asfimport]

Tony Yan:
Thanks for the fix.

[asfimport]

Alex Levenson / @isnotinvain:
Issue resolved by pull request 220
#220