[fix][broker] Duplicate ByteBuffer when Caching Backlogged Consumers

[michaeljmarshall]

Fixes #16979

Motivation

#12258 introduced caching for backlogged consumers. When caching the entry, it is important to duplicate the ByteBuffer so that the reader index is not shared. The current code has a race condition where the ByteBuffer reference in the cache is shared with the dispatcher. When another consumer reads from the cache, the cache calls duplicate() on the shared ByteBuffer, which copies the current reader index, which might not be 0 if the original dispatcher read data from the ByteBuffer.

Note: it seems like the caching insert method creates (or recycles) more EntryImpl instances than is really necessary. Changing that is outside this PR's scope, so I am going to leave it as is.

Modifications

• Create a new Entry before inserting it into the cache.
• Add a new test to the EntryCacheTest. The test fails before this change and passes after it.
• Update the EntryCacheTest mocking so that it returns unique entries when mocking reads from the bookkeeper. Before, all returned LedgerEntry objects had ledgerId 0 and entryId 0, which messed with the caching for the new test.

Verifying this change

This change includes a test that failed before the PR and passes after it.

Documentation

• doc-not-needed

[mattisonchao]

Nice catch!

[mattisonchao]

Small suggestion:

Could we make the insert method ensure it? It can avoid the same problem when anybody inserts it in the future.
Maybe we can make cachedData = entry.getDataBuffer().retain(); use retainedDuplicate().
Or change the insert method to insert(EntryImpl entry, boolean duplicateDataBuffer) and the duplicateDataBuffer default is true. you can set it to false when you ensure you never change the original ByteBuffer. (Just quick think and still need deep thinking).

The advantage is that we make this method more fault-tolerant and don't need create another EntryImpl

[michaeljmarshall]

Small suggestion:

Could we make the insert method ensure it? It can avoid the same problem when anybody inserts it in the future. Maybe we can make cachedData = entry.getDataBuffer().retain(); use retainedDuplicate(). Or change the insert method to insert(EntryImpl entry, boolean duplicateDataBuffer) and the duplicateDataBuffer default is true. you can set it to false when you ensure you never change the original ByteBuffer. (Just quick think and still need deep thinking).

The advantage is that we make this method more fault-tolerant and don't need create another EntryImpl

@mattisonchao - I didn't do it this way because the write path for tail read caching does not need the duplicate ByteBuffer created. That path technically doesn't even need an Entry, but it creates one because insert only takes an Entry. It seems like we should consider expanding/improving the EntryCache interface to meet these two different use cases. I agree that we should think about simplifying the entry reference management. I don't have benchmarks to show how much it costs to create an Entry unnecessarily, but this part of the code base is highly optimized, so I would prefer not to introduce unnecessary duplication if we can avoid it.

I am going to merge this as is so that it does not hold up the 2.11 release at all. We can definitely continue this discussion, especially because we have the PIP on improved caching getting implemented right now.