0xffff-zhiyan commented on code in PR #22678:
URL: https://github.com/apache/kafka/pull/22678#discussion_r3554209078
##########
raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java:
##########
@@ -461,4 +466,80 @@ 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.times(1));
+ }
+
+ 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();
+
+ List<BufferSupplier> suppliers = new ArrayList<>();
Review Comment:
because two `suppliers` were created in this scenario. We need to capture
both instances and verify each of them is closed.
--
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]
