[jira] [Work logged] (BEAM-14187) NullPointerException or IllegalStateException at IsmReaderImpl in Dataflow

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/BEAM-14187?focusedWorklogId=754281&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-754281 ]

ASF GitHub Bot logged work on BEAM-14187:
-----------------------------------------

                Author: ASF GitHub Bot
            Created on: 07/Apr/22 18:24
            Start Date: 07/Apr/22 18:24
    Worklog Time Spent: 10m 
      Work Description: lukecwik commented on code in PR #17201:
URL: https://github.com/apache/beam/pull/17201#discussion_r845437349


##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IsmReaderImpl.java:
##########
@@ -370,7 +373,7 @@ boolean bloomFilterMightContain(RandomAccessData keyBytes) {
     position(rawChannel, footer.getBloomFilterPosition());
     bloomFilter = ScalableBloomFilterCoder.of().decode(Channels.newInputStream(rawChannel));
 
-    indexPerShard = new HashMap<>();
+    indexPerShard = new ConcurrentHashMap<>();

Review Comment:
   Note that initializeBloomFilterAndIndexPerShard is synchronized which is the only method that mutates the indexPerShard variable. Since initializeBloomFilterAndIndexPerShard is invoked within overKeyComponents that means there is a memory barrier between the thread that created the indexPerShard and other threads ensuring that the object is safely published across threads.
   
   Also, the [Javadoc for HashMap](https://docs.oracle.com/javase/8/docs/api/java/util/HashMap.html) says that concurrent reads are ok, its concurrent modifications that require synchronization:
   "If multiple threads access a hash map concurrently, and at least one of the threads modifies the map structurally, it must be synchronized externally."





Issue Time Tracking
-------------------

    Worklog Id:     (was: 754281)
    Time Spent: 40m  (was: 0.5h)

> NullPointerException or IllegalStateException at IsmReaderImpl in Dataflow
> --------------------------------------------------------------------------
>
>                 Key: BEAM-14187
>                 URL: https://issues.apache.org/jira/browse/BEAM-14187
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-dataflow
>            Reporter: Minbo Bae
>            Priority: P2
>          Time Spent: 40m
>  Remaining Estimate: 0h
>
> h6. Problem
> Dataflow Java batch jobs with large side input intermittently throws {{NullPointerException}} or {{{}IllegalStateException{}}}.
>  * [NullPointerException|https://gist.githubusercontent.com/baeminbo/459e283eadbc7752c9f23616b52d958a/raw/f0480b8eaff590fb3f3ae2ab98ddce7dd3b4a237/npe.png] happens at [IsmReaderImpl.overKeyComponents|https://github.com/apache/beam/blob/v2.37.0/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IsmReaderImpl.java#L217]:
>  * [IllegalStateException|https://gist.githubusercontent.com/baeminbo/459e283eadbc7752c9f23616b52d958a/raw/f0480b8eaff590fb3f3ae2ab98ddce7dd3b4a237/IllegalStateException.png] happens at [IsmReaderImpl. initializeForKeyedRead |https://github.com/apache/beam/blob/v2.37.0/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IsmReaderImpl.java#L500].
> (all error logs in the Dataflow job is [here|https://gist.githubusercontent.com/baeminbo/459e283eadbc7752c9f23616b52d958a/raw/f0480b8eaff590fb3f3ae2ab98ddce7dd3b4a237/downloaded-logs-20220327-171955.json].)
> h6. Hypothesis
> The {{initializeForKeyedRead}} is not synchronized. Multiple threads can enter the method so that initialize the index for the same shard and update {{indexPerShard}} without synchronization. And, the {{overKeyComponents}} also accesses {{indexPerShard}} without synchronization. As {{indexPerShard}} is just a {{HashMap}} which is not thread-safe, it can cause {{NullPointerException}} and {{IllegalStateException}} above.
> h6. Suggestion
> I think it can fix this issue if we change the type of {{indexPerShard}} to a thread-safe map (e.g. {{ConcurrentHashMap}}).



--
This message was sent by Atlassian Jira
(v8.20.1#820001)