KAFKA-14491: [3/N] Add logical key value segments

[vcrfxia]

Today's KeyValueSegments create a new RocksDB instance for each KeyValueSegment. This PR introduces an analogous LogicalKeyValueSegments implementation, with corresponding LogicalKeyValueSegment, which shares a single physical RocksDB instance across all "logical" segments. This will be used for the RocksDB versioned store implementation proposed in KIP-889.

Committer Checklist (excluded from commit message)

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

[mjsax]

If the store was never open, it seems it's still safe to call closeOpenIterators and it should just be an empty list? -- Could we inline the code into close() directly?

[vcrfxia]

Sure, I don't feel strongly so I made the change. Besides guarding against closing a segment which was never opened, the usage of open also guarded against closing the same segment twice. I've inlined closeOpenIterators() and accounted for this by clearing openIterators after it's copied.

[mjsax]

How large is the overhead to call wrap() (besides that it create a new object, what does it do?)

We pass in Bytes prefix in the constructor and seem if we keep a reference, we could just return it (without the need to unwrap in the constructor and re-wrap here?

[vcrfxia]

wrap() just creates the new object (after performing a null check) so it's very lightweight.

It's more convenient to keep prefix as byte[] than Bytes because all the other operations require byte[] rather than Bytes. If we really wanted we could keep both (one copy as byte[] and another as Bytes) but that feels like overkill.

[vcrfxia]

Thanks @mjsax for the speedy review! Addressed your comments in the latest commit.

[vcrfxia]

Sure, I don't feel strongly so I made the change. Besides guarding against closing a segment which was never opened, the usage of open also guarded against closing the same segment twice. I've inlined closeOpenIterators() and accounted for this by clearing openIterators after it's copied.

[vcrfxia]

wrap() just creates the new object (after performing a null check) so it's very lightweight.

It's more convenient to keep prefix as byte[] than Bytes because all the other operations require byte[] rather than Bytes. If we really wanted we could keep both (one copy as byte[] and another as Bytes) but that feels like overkill.

[mjsax]

Seems some of you newly added tests fail. Can you have a look.

[vcrfxia]

Thanks. Needed to update the test for the latest changes which now set isOpen = true always. Fixed now.

[mjsax]

Checkstyle error:

> Task :streams:checkstyleTest

[2023-01-26T23:03:19.137Z] [ant:checkstyle] [ERROR] /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-13143/streams/src/test/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegmentsTest.java:23:15: Unused import - org.junit.Assert.assertFalse. [UnusedImports]

[mjsax]

Merged the other PR -- can you rebase this one?

[mjsax]

Just some minor follow up on tests.

[mjsax]

Should we also get on the physical store to see if the logic works as expected? (Also for other tests)

[vcrfxia]

I was on the fence about this because it requires testing the internals of the class (i.e., specifically how the segment prefixes are serialized) rather than just the public-facing methods. In the end I opted to test indirectly instead, by inserting the same keys into different segments and checking that their values do not collide.

If you prefer checking the contents of the physical store itself, I can make the update.

[mjsax]

I see your point, but the test does not really achieve this, as we put the same data into both segment? To test "segment isolation" we would need to put 4 different record (2 per segment) and test both positive (put on s1 allows use to get on s1) and negative (put on s1, does not allow get on s2 to see the data)?

Might apply to other tests, too?

[vcrfxia]

Ah good point. That's definitely a gap in shouldPut() and shouldPutAll(). All of the other tests are already set up in a way that they fail if segments are not properly isolated from each other. Just pushed a fix to the two tests which didn't ensure that, and some minor cleanup to a few of the other tests.

[mjsax]

Should we test different ranges? All lower and upper bound null/not-null combination?

[vcrfxia]

Heh, this additional test coverage caught a bug. Pushed a fix in the latest commit.

[mjsax]

Should we call getOrCreateSegment instead? Otherwise we mainly test the logic of AbstractSegments ?

[vcrfxia]

See above.

[mjsax]

Sound like we test AbstractSegments logic here -- do we need to do this?

[vcrfxia]

You're right that these tests are testing logic from AbstractSegments and not anything specific about LogicalKeyValueSegments. The thing is, AbstractSegments doesn't have its own test file at the moment (I assume because it's abstract). If you think it's worth it, I can remove these tests from here and also from KeyValueSegmentsTest.java, and create a dummy AbstractSegments implementation to add an AbstractSegmentsTest.java. I'd like to do that as a follow-up PR instead of as part of this change, though.

(Also, for this specific test, I would like to have it here because I plan to refactor the cleanup logic in AbstractSegments in a follow-up PR. The current approach (cleanup as part of getOrCreateSegmentIfLive()) is not very efficient for the versioned store use case because this method is called multiple times during a single put operation. It will be better to only perform cleanup once per put.)

[mjsax]

If you think it's worth it, I can remove these tests from here and also from KeyValueSegmentsTest.java, and create a dummy AbstractSegments implementation to add an AbstractSegmentsTest.java. I'd like to do that as a follow-up PR instead of as part of this change, though.

Sounds cleaner to me. And yes, follow up PR is preferable.

[mjsax]

Ab above

[mjsax]

ab above

[vcrfxia]

Thanks for your review, @mjsax ! Responded to your comments inline. Will push a commit with the latest changes soon.

[vcrfxia]

I was on the fence about this because it requires testing the internals of the class (i.e., specifically how the segment prefixes are serialized) rather than just the public-facing methods. In the end I opted to test indirectly instead, by inserting the same keys into different segments and checking that their values do not collide.

If you prefer checking the contents of the physical store itself, I can make the update.

[vcrfxia]

You're right that these tests are testing logic from AbstractSegments and not anything specific about LogicalKeyValueSegments. The thing is, AbstractSegments doesn't have its own test file at the moment (I assume because it's abstract). If you think it's worth it, I can remove these tests from here and also from KeyValueSegmentsTest.java, and create a dummy AbstractSegments implementation to add an AbstractSegmentsTest.java. I'd like to do that as a follow-up PR instead of as part of this change, though.

(Also, for this specific test, I would like to have it here because I plan to refactor the cleanup logic in AbstractSegments in a follow-up PR. The current approach (cleanup as part of getOrCreateSegmentIfLive()) is not very efficient for the versioned store use case because this method is called multiple times during a single put operation. It will be better to only perform cleanup once per put.)