KAFKA-7912: Support concurrent access in InMemoryKeyValueStore

[ableegoldman]

Previously the InMemoryKeyValue store would throw a ConcurrentModificationException if the store was modified beneath an open iterator. The TreeMap implementation was swapped with a ConcurrentSkipListMap for similar performance while supporting concurrent access.

Added one test to AbstractKeyValueStoreTest, no existing tests caught this.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[ableegoldman]

No need to copy the entire underlying store, just return an iterator on the existing map

[ableegoldman]

@guozhangwang @vvcephei @bbejeck @mjsax

[bbejeck]

\cc @vvcephei for review

[bbejeck]

Java 8 failed kafka.server.ListOffsetsRequestTest.testResponseIncludesLeaderEpoch
Java 11 failed

kafka.api.SaslSslAdminClientIntegrationTest.testMinimumRequestTimeouts
org.apache.kafka.streams.integration.RegexSourceIntegrationTest.testRegexMatchesTopicsAWhenDeleted

retest this please

[bbejeck]

Thanks, @ableegoldman good catch.

Overall it looks good to me, but this got me thinking.

Since we only have a store per task and a StreamThread only operates on one task at a time (even if it is assigned multiple tasks) could we hit this point in a streams application under regular circumstances? I guess the risk comes from the fact that someone could open an iterator and hold on to it without closing and then during further processing the map is updated and since the iterator is a view back to the original map, using the iterator again results in the exception

I'm not suggesting we don't merge this PR, as IMHO it's an improvement, it's got me thinking about our store access patterns.

[ableegoldman]

It does seem like we're adding "some" overhead here to deal with edge cases of weird behavior for IQ only, since as you say we shouldn't hit this problem under normal operation. Might it be worth implementing two versions of the underlying stores, an optimized one for normal Streams and a "safe" one for IQ?

I'll try to investigate how much of an overhead we're really introducing here, I suspect it's not terrible for the key-value store but might be less tolerable for the window store (or future session store) since I believe SkipList performance would degrade as we continually delete from the beginning as things expire..?

[bbejeck]

Might it be worth implementing two versions of the underlying stores, an optimized one for normal Streams and a "safe" one for IQ?

I'm leaning towards no, if there is any additional overhead I don't think it would be that much and returning thread-safe iterators for IQ would outweigh the cost of maintaining two separate types.

Additionally, we have ReadOnlyXXXStore interfaces that need to provide thread-safe access, so I think what you have here will fit in well.

[bbejeck]

Java 11 failed with

kafka.api.DelegationTokenEndToEndAuthorizationTest.testNoConsumeWithoutDescribeAclViaSubscribe
kafka.server.DynamicBrokerReconfigurationTest.testDefaultTopicConfig
kafka.server.RequestQuotaTest.testUnauthorizedThrottle

Java 8 failed with kafka.api.AdminClientIntegrationTest.testMinimumRequestTimeouts

retest this please

[guozhangwang]

Good point about performance degradation of SkipList if most of the modifications are at the top -- well depending on how clever J8's SkipList implementation is, as they can go as fancy as this: https://arxiv.org/pdf/1805.04794.pdf (related works introduced skiplist).

But I think for now we can still afford for the overhead, even for windowed in-memory store: if we realized later in production that it is really bad we can come back to it as well.

[guozhangwang]

LGTM (java8 will not succeed so we can ignore atm).

[bbejeck]

LGTM

[guozhangwang]

Also confirmed that the added test will fail without the other change.

Merged to trunk, thanks @ableegoldman !!