KEFCore fails some tests in net8.0 and JDK 21

[masesdevelopers]

Within https://github.com/masesgroup/KEFCore/actions/runs/9655461268 many tests fails. Here the summary:

• .NET: net8.0
• OS
  • Windows
    • JDK: Corretto 21
  • Linux
    • Oracle 21
    • Zulu 21
    • Temurin 21
    • Microsoft 21
  • MacOS 13 and MacOS 14 (macos-latest)
    • Corretto 21
    • Microsoft 21
    • Temurin 21
    • Oracle 21
    • Zulu 21

The errors occurs always in net8.0, and sparsely for many JDK 21, when the array of bytes of key related to the Kafka record is retrieved from the JVM.
However the problem is raised in many way within KNet code, so the issue shall be solved there.

Originally posted by @masesdevelopers in masesgroup/KEFCore#243 (comment)

[masesdevelopers]

#480 is still open, maybe specific tests will be added there

[masesdevelopers]

Check this comment for some issues found

[masesdevelopers]

Tests still have some issues reported in https://github.com/masesgroup/KEFCore/actions/runs/9703496875; some of them can be mitigated with masesgroup/KEFCore#243 (comment) since the broker/zookeeper fails in windows, however someone else seems specific to JDK 21 and especially regards memory management: e.g. Arithmetic overflow reported on byte array saved in previous steps, so something happens when memory is read, or wrote.

[masesdevelopers]

Tests still have some issues reported in https://github.com/masesgroup/KEFCore/actions/runs/9703496875; some of them can be mitigated with masesgroup/KEFCore#243 (comment) since the broker/zookeeper fails in windows, however someone else seems specific to JDK 21 and especially regards memory management: e.g. Arithmetic overflow reported on byte array saved in previous steps, so something happens when memory is read, or wrote.

Finally seems the issue was identified and it is related to the access of information stored in the final fields (key, value) of org.apache.kafka.streams.KeyValue<K, V>, however it is not clear at all the reason why the, in JDK 21, the access to the memory creates so many issues: sometime the length read is negative (and indeed .NET reports Arithmetic overflow), sometime is very high creating an access violation since the code tries to access an inaccessible memory, sometime the length is too short, sometime the memory accessed during the copy is not the one of the expected array.

The solution identified is to create a new class, named org.mases.knet.developed.streams.KeyValueSupport<K, V>, which accept in input an instance of org.apache.kafka.streams.KeyValue<K, V> and give access to the key and value fields with methods:

public class KeyValueSupport<K, V> {
    org.apache.kafka.streams.KeyValue<K, V> _innerKV;

    public static <K, V> org.apache.kafka.streams.KeyValue<K, V> toKeyValue(KeyValueSupport<K, V> kvs) {
        return new org.apache.kafka.streams.KeyValue<K, V>(kvs.getKey(), kvs.getValue());
    }

    public KeyValueSupport(org.apache.kafka.streams.KeyValue<K, V> innerKV) {
        _innerKV = innerKV;
    }

    public K getKey() {
        return _innerKV.key;
    }

    public V getValue() {
        return _innerKV.value;
    }
}

Using this approach seems that the memory is accessed in the right way and the tests done locally does not report the issues described before.
Probably a better approach is to create a new class which implements org.apache.kafka.streams.state.KeyValueIterator and iterates managing org.mases.knet.developed.streams.KeyValueSupport<K, V>, till now leave this approach as a new option.