KAFKA-7534: Error in flush calling close may prevent underlying store from closing

[bbejeck]

Calling the CachingKeyValueStore#close() method first calls CachingKeyValueStore.flush(). If there is an exception thrown during the flush call, the underlying store is not closed. Subsequently, another task can't open the RocksDB store and receives a No locks available exception.

I added a unit test that fails with the proposed fix.

Committer Checklist (excluded from commit message)

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

[bbejeck]

ping @mjsax @vvcephei for reviews

[mjsax]

Should we have a nested try-catch to make use cache.close() is called, too? Not sure how critical it is to ensure that the cache is closed?

[bbejeck]

I had that, but I took it out for the same reason thinking it's not important to ensure the cache is closed. But thinking about it some more, it doesn't hurt anything, so I'll put it back.

[bbejeck]

updated this

[bbejeck]

Java 11 build timed out.

retest this please

[bbejeck]

failure unrelated

retest this please

[vvcephei]

LGTM. Thanks, @bbejeck .

[bbejeck]

retest this please

[bbejeck]

@omkreddy can we please get this PR into 2.0.1?

[bbejeck]

Both builds timed out after 3 hours

retest this please

[mjsax]

Java11: https://builds.apache.org/job/kafka-pr-jdk11-scala2.12/293/testReport/junit/kafka.log/LogCleanerIntegrationTest/testMarksPartitionsAsOfflineAndPopulatesUncleanableMetrics/

kafka.log.LogCleanerIntegrationTest.testMarksPartitionsAsOfflineAndPopulatesUncleanableMetrics failed with

org.junit.runners.model.TestTimedOutException: test timed out after 15000 milliseconds

Merging. Java8 is green. \cc @vvcephei

[mjsax]

Merged to trunk and cherry-picked to 2.1, 2.0, 1.1, and 1.0 branch. Will mark with fixed version 2.2.0 for now -- if new RCs are rolled, we should add other fixed versions accordingly.