KAFKA-10409: Refactor Kakfa Streams RocksDB Iterators

[fonsdant]

Key refactorings:

• Use RocksDBDualCFRangeIterator for DualColumnFamilyAccessor#all, since it iterate over all when passing null as from and to key ranges;
• Create helper methods to:
  • loadNextKeys();
  • fetchNextKeyValue();
  • handleWithTimestampOnly();
  • handleNoTimestampOnly();
  • compareAndHandleKeys();
  • check if some key-value isInRange(); and
  • ensureOpen().
• Rewrite methods with new helper methods:
  • makeNext();
  • next(); and
  • hasNext().
• Improve members naming with:
  • noTimestampNext;
  • withTimestampNext;
  • noTimestampIterator; and
  • withTimestampIterator.
• Sort members and method arguments alphanumerically;
• Add documentation.

[github-actions]

A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the triage label
should be removed to prevent this automation from happening again.

[fonsdant]

Hi, @ableegoldman! Could you help me with this PR? :)

[ableegoldman]

Yep! Will take a look in the next day or so 🙂

[agavra]

thanks @fonsdant this is a great cleanup! I left some minor suggestions inline, I'm not super familiar with this part of the code so feel free to take it or leave it

[agavra]

what does this mean? doesn't that just mean it's not thread safe?

[fonsdant]

Updated.

[agavra]

existing test coverage for this class is below average (e.g. there's not peekNextKey in the test) - could you add that?

cc @ableegoldman it may be good to refactor the existing tests alongside this one so that instead of one really big test we had individual tests for each piece of functionality so it's easier to confirm the test coverage is good

[fonsdant]

In progress.

[agavra]

There seems to be a functionality change. Not sure if this is intentional or not but it looks like we drop checking the condition where the iterators are not valid. (In other words, there's a branch where after loadNextKeys(), noTimestampNext is still null but iterator.isValid().

Not sure what might cause this condition.

The old code:

            if (nextNoTimestamp == null && iterNoTimestamp.isValid()) {
                nextNoTimestamp = iterNoTimestamp.key();
            }

            if (nextWithTimestamp == null && iterWithTimestamp.isValid()) {
                nextWithTimestamp = iterWithTimestamp.key();
            }
            if (nextNoTimestamp == null && !iterNoTimestamp.isValid()) {
                if (nextWithTimestamp == null && !iterWithTimestamp.isValid()) {
                    return allDone();

[fonsdant]

Although this branch is not explicit, I think it is covered. In the old code, allDone is only called if both nextTimestamp are null and both are invalid. So this is already checked on previous branches (in the new code, on loadNextKeys). That is, if both next are null, in loadNextKeys: 1) if they are valid, they will get next keys and will no longer be null (and will not go into allDone when loadNextKeys is finished since they have become different from null); 2) if they are invalid, they will remain null and will go into allDone after loadNextKeys is finished.

[agavra]

Suggested change

	return forward
	? comparison < 0 || (toInclusive && comparison == 0)
	: comparison > 0 || (toInclusive && comparison == 0);
	return (toInclusive && comparison == 0) || (forward ? comparison < 0 : comparison > 0);

[fonsdant]

Updated.

[agavra]

generally it's better for constructors not to have side effects - otherwise it's possible that if something throws an exception during instantiation the close method won't be called.

this doesn't seem to be a regression (the old code did that as well for the RangeIterator) so consider this comment an optional suggestion

[fonsdant]

Updated.

[fonsdant]

@agavra, thanks for reviewing! I have pushed some commits :)

[agavra]

This LGTM once the tests are in :) will make it easier to convince me that there's no functionality changes. Thanks for addressing the comments.

[ableegoldman]

Took a quick look and I think it looks good, but as Almog mentioned the existing test coverage is seriously lacking so it'll be great to fill in some the tests here. I'll do a closer pass once you push the new tests 🙂

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[fonsdant]

@agavra, sorry for the delay, I am resuming the work. How could I check the coverage report for streams only? According on readme, I can run gradle streams:reportCoverage, but when I do this, I get: * What went wrong: Cannot locate tasks that match 'streams:reportCoverage' as task 'reportCoverage' not found in project ':streams'.

P.S.: I was missing -PenableTestCoverage=true, I got it :)