[FLINK-38783][runtime] Improved TieredStorageResourceRegistry Thread-Safety/Concurrency

[rionmonster]

What is the purpose of the change

This pull request addresses the issue detailed in FLINK-38783 which detailed how the registration process within TieredStorageResourceRegistry was not properly handling concurrent operations and as such could throw a ConcurrentModificationException under concurrent load.

Brief change log

• Replaced the previous registeredResources general-purpose hash map (Map<..., List<...>>) with corresponding thread-safe structures (ConcurrentHashMap<..., CopyOnWriteArrayList<...>>) to better handle concurrent operations.

Verifying this change

This change added a series of tests in TieredStorageResourceRegistryTest to originally reproduce the issue and later confirm the fix worked as expected including:

• testConcurrentRegisterResource to test concurrent resource registration across separate threads (10 total with same owner/identifier)
• testConcurrentRegisterResourceWithDifferentOwners to test concurrent registration across separate threads (10 total with different owners/identifiers)
• testConcurrentRegisterAndClear to test concurrent registration and clearing across separate threads

Example Tests

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): no
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: no/don't know
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

[rionmonster]

CC: @TanYuxin-tyx for review (please feel free to tag someone else). You seemed to be the most appropriate given the history.

[flinkbot]

CI report:

• 448adde Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[rionmonster]

@flinkbot run azure

[davidradl]

I am curious, why do we need the CopyOnWriteArrayList, is the introduction of ConcurrentHashMap not enough to solve this?

[rionmonster]

That's a good question!

I thought the same when I initially applied the fix (e.g., only swapped out the external map for its thread-safe brethren), however realized the tests that were added would still fail.

The ConcurrentHashMap handles the thread-safety for the map operations but not for the internal values within the map. This makes it possible to have multiple separate threads acting upon the non thread-safe list, which can lead to some inconsistency:

registeredResources
      .computeIfAbsent(owner, (ignore) -> new ArrayList<>())
      // Concurrent callers could be working with the same thread-safe map, but
      // the underlying list is not thread-safe
      .add(tieredStorageResource);

Without the extra thread-safety on the list, many of the existing tests can fail with ConcurrentModificationException, NullPointerException, and lost entries (which testConcurrentRegisterWithSameIdentifier specifically checks for). Making the swap to the CopyOnWriteArrayList (or some other thread-safe collection like Collections.synchronizedList()) makes the behavior consistent.