NIFI-7549 Adding Hazelcast based DistributedMapCacheClient support

[simonbence]

NIFI-7549

The PR contains my proposal of Hazelcast support for DistributedMapCacheClient. In general, I followed the patterns I found in the existing implementations, for the cases were not explicitly documented the behaviour follows them, mainly the ones were added with the feature itself (I considered them the most relevant and accurate implementations)

As for the organisation of the implementation, I did split the feature into three "layers". The package structure follows this as well. In the bottom, there is the HazelcastCache, and the implementation. This layer is responsible to directly communicate with the Hazelcast (via a provided connection) and hide the details of the used data structure. The current implementation is based on IMap, but there is the possibility to change or extend this. Also, the map-like data structure's interface is heavily changed between Hazelcast 3.x and 4.x. In case if the support would be needed for older implementations, wrapping the logic could help to avoid sprawl of the changes.

The layer above is the "cache manager" (HazelcastCacheManager). This is responsible to create the cache instances and maintain the connection. Currently there are two implementation: one which starts an embedded Hazelcast for easy usage and one which connects to a Hazelcast cluster running outside NiFi. The embedded provides a limited capability for configuration, but it could serve effectively as local cache. Also it supports clustering the embedded Hazelcast instances. The "standalone" could joint to any non-enterprise Hazelcast. Note: I looked after how to connect with secured Hazelcast, but as I found it is part of the enterprise package. For now, it was not part of my intent to support that. This layer should hide all Hazelcast specific interface or implementation.

The top layer is the actual DistributedMapCacheClient implementation. Depends on both the bottom ones, as the manager is needed for acquiring the cache which it works with. All the NiFi specific logic is within this. AtomicDistributedMapCacheClient methods are supported. The revision handling comes in with this is general for all the entries. A long-based version is attached to all the entries.

Thank you for submitting a contribution to Apache NiFi.

Please provide a short description of the PR here:

Description of PR

Enables X functionality; fixes bug NIFI-YYYY.

In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:

For all changes:

• Is there a JIRA ticket associated with this PR? Is it referenced
  in the commit message?

• Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.

• Has your PR been rebased against the latest commit within the target branch (typically main)?

• Is your initial contribution a single, squashed commit? Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not squash or use --force when pushing to allow for clean monitoring of changes.

For code changes:

• Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
• Have you written or updated unit tests to verify your changes?
• Have you verified that the full build is successful on JDK 8?
• Have you verified that the full build is successful on JDK 11?
• If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under ASF 2.0?
• If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
• If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
• If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?

For documentation related changes:

• Have you ensured that format looks appropriate for the output in which it is rendered?

Note:

Please ensure that once the PR is submitted, you check GitHub Actions CI for build issues and submit an update to your PR as soon as possible.

[kevdoran]

Thanks @simonbence! At a glance, looks very well thought out and designed. Will review...

[turcsanyip]

Licensing info is missing from the nar modules. LICENSE / NOTICE files need to be added with entries to the referenced libraries (eg. hazelcast).

[tpalfy]

Looks good, I'd just add a test for serialization/deserialization in HazelcastMapCacheClientTest:

    @Test
    public void testSerialization() throws Exception {
        // GIVEN
        Long key = 1L;
        Double value = 1.2;

        Serializer<Long> keySerializer = (x, output) -> output.write(x.toString().getBytes(StandardCharsets.UTF_8));
        Serializer<Double> valueSerializer = (x, output) -> output.write(x.toString().getBytes(StandardCharsets.UTF_8));

        Deserializer<Double> valueDeserializer = input -> Double.valueOf(new String(input, StandardCharsets.UTF_8));
        testSubject.put(key, value, keySerializer, valueSerializer);

        // WHEN
        Double actual = testSubject.get(key, keySerializer, valueDeserializer);

        // THEN
        assertEquals(value, actual);
    }

[turcsanyip]

@simonbence Thanks for the fixes so far!
I found some more (mostly minor) issues related to the property validation. Please check them too.

[tpalfy]

LGTM+1

[kevdoran]

Great work! The layered design which allows for both internal and external Hazelcast is really nice. For folks familiar with Hazelcast, this is very intuitive to use, and will be a nice additional option for NiFi distributed caches.

The only things I spotted would be perhaps exposing more configurability for the embedded HZ cluster, such as Split Brain Protection. Given that a reasonable alternative to this in the current PR is using an external HZ cluster which would provide complete control of the HZ config, I do not view this as a blocker by any means. There are however, a couple places that I think exception handling or retry logic could be added. I have called out one such example for IMap usage when split brain protection is enabled.

Overall, great design, implementation and documentation! Nice work! +1

[kevdoran]

Excellent documentation!

[kevdoran]

If Hazelcast 4 is similar to 3 in this regard, then I believe there are some other runtime-exceptions that IMap datastructures can throw. One in particular is a SplitBrainProtectionException, which can occur on an external HZ cluster if nodes lose connectivity to each other, such as unwanted network partitioning. It might be worth adding handling for that, usually the solution would be a configurable number of retries as the same operation can succeed if called to a HZ instance that is part of the majority/quorum above the split brain protection threshold.

[turcsanyip]

+1 LGTM

Tested with different cache configs on NiFi + Hazelcast clusters.

@simonbence Thanks for implementing this new component.
@tpalfy, @kevdoran Thanks for the reviews.

Merging to main...