0xffff-zhiyan commented on code in PR #22678:
URL: https://github.com/apache/kafka/pull/22678#discussion_r3554122023
##########
raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java:
##########
@@ -461,4 +463,74 @@ void testTrimPrefixTo() {
assertEquals(Optional.of(voterSet),
partitionState.voterSetAtOffset(voterSetOffset));
assertEquals(4, getNumberOfVoters(metrics).metricValue());
}
+
+ @Test
+ void testBufferSupplierCreatedAndClosedOnLogRead() {
+ Metrics metrics = new Metrics();
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ ExternalKRaftMetrics externalMetrics =
Mockito.mock(ExternalKRaftMetrics.class);
+ MockLog log = buildLog();
+ VoterSet staticVoterSet =
VoterSetTest.voterSet(VoterSetTest.voterMap(IntStream.of(1, 2, 3), true));
+ int epoch = 1;
+
+ KRaftControlRecordStateMachine partitionState =
buildPartitionListener(log, staticVoterSet, raftMetrics, externalMetrics);
+
+ // Append a control record so that the log has something to read.
+ VoterSet voterSet =
VoterSetTest.voterSet(VoterSetTest.voterMap(IntStream.of(4, 5, 6), true));
+ log.appendAsLeader(
+ MemoryRecords.withVotersRecord(
+ log.endOffset().offset(),
+ 0,
+ epoch,
+ BufferSupplier.NO_CACHING.get(300),
+ voterSet.toVotersRecord((short) 0)
+ ),
+ epoch
+ );
+
+ BufferSupplier supplier = spy(new
BufferSupplier.GrowableBufferSupplier());
+ try (MockedStatic<BufferSupplier> bufferSupplierMock =
mockStatic(BufferSupplier.class)) {
+
bufferSupplierMock.when(BufferSupplier::create).thenReturn(supplier);
+
+ partitionState.updateState();
+
+ // The log read creates a BufferSupplier and closes it via
try-with-resources.
+ bufferSupplierMock.verify(BufferSupplier::create,
Mockito.atLeastOnce());
+ }
+
+ Mockito.verify(supplier).close();
+ }
+
+ @Test
+ void testBufferSupplierCreatedOnSnapshotRead() {
+ Metrics metrics = new Metrics();
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ ExternalKRaftMetrics externalMetrics =
Mockito.mock(ExternalKRaftMetrics.class);
+ MockLog log = buildLog();
+ VoterSet staticVoterSet =
VoterSetTest.voterSet(VoterSetTest.voterMap(IntStream.of(1, 2, 3), true));
+ int epoch = 1;
+
+ KRaftControlRecordStateMachine partitionState =
buildPartitionListener(log, staticVoterSet, raftMetrics, externalMetrics);
+
+ // Create a snapshot with control records so that the snapshot has
something to read.
+ KRaftVersion kraftVersion = KRaftVersion.KRAFT_VERSION_1;
+ VoterSet voterSet =
VoterSetTest.voterSet(VoterSetTest.voterMap(IntStream.of(4, 5, 6), true));
+ RecordsSnapshotWriter.Builder builder = new
RecordsSnapshotWriter.Builder()
+ .setRawSnapshotWriter(log.createNewSnapshotUnchecked(new
OffsetAndEpoch(10, epoch)).get())
+ .setKraftVersion(kraftVersion)
+ .setVoterSet(Optional.of(voterSet));
+ try (RecordsSnapshotWriter<?> writer = builder.build(STRING_SERDE)) {
+ writer.freeze();
+ }
+ log.truncateToLatestSnapshot();
+
+ try (MockedStatic<BufferSupplier> bufferSupplierMock =
mockStatic(BufferSupplier.class)) {
+
bufferSupplierMock.when(BufferSupplier::create).thenAnswer(invocation -> new
BufferSupplier.GrowableBufferSupplier());
+
+ partitionState.updateState();
+
+ // The snapshot read creates a BufferSupplier.
+ bufferSupplierMock.verify(BufferSupplier::create,
Mockito.atLeastOnce());
+ }
Review Comment:
The supplier here is the spied `GrowableBufferSupplier` instance that's
injected via the mocked create(). So `verify(supplier).close()` is already
verifying `close()` on that instance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
