Repository: apache/kafka
File: coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeTest.java (Java)

Code before:
            new MockCoordinatorShard.RecordAndMetadata(3, records.get(3))
        ), ctx.coordinator.coordinator().fullRecords());
        assertEquals(List.of(
            records(secondBatchTimestamp, records.subList(0, 3)),
            records(secondBatchTimestamp, records.subList(3, 4))
        ), writer.entries(TP));
        verify(runtimeMetrics, times(2)).recordFlushTime(10);

        // Commit and verify that writes are completed.
        writer.commit(TP);
        assertTrue(write1.isDone());
        assertTrue(write2.isDone());
        assertTrue(write3.isDone());
        assertEquals(4L, ctx.coordinator.lastCommittedOffset());
        assertEquals("response1", write1.get(5, TimeUnit.SECONDS));
        assertEquals("response2", write2.get(5, TimeUnit.SECONDS));
        assertEquals("response3", write3.get(5, TimeUnit.SECONDS));
    }

    @Test
    public void testLargeCompressibleRecordTriggersFlushAndSucceeds() throws Exception {
        MockTimer timer = new MockTimer();

        // Writer sleeps for 10ms before appending records.
        MockPartitionWriter writer = new MockPartitionWriter(timer.time(), Integer.MAX_VALUE, false);
        CoordinatorRuntimeMetrics runtimeMetrics = mock(CoordinatorRuntimeMetrics.class);
        Compression compression = Compression.gzip().build();

        CoordinatorRuntime<MockCoordinatorShard, String> runtime =
            new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
                .withTime(timer.time())
                .withTimer(timer)
                .withDefaultWriteTimeOut(Duration.ofMillis(20))
                .withLoader(new MockCoordinatorLoader())
                .withEventProcessor(new DirectEventProcessor())
                .withPartitionWriter(writer)
                .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
                .withCoordinatorRuntimeMetrics(runtimeMetrics)
                .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
                .withCompression(compression)
                .withSerializer(new StringSerializer())
                .withAppendLingerMs(10)
                .withExecutorService(mock(ExecutorService.class))
                .build();

        // Schedule the loading.
        runtime.scheduleLoadOperation(TP, 10);

        // Verify the initial state.
        CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
        assertNull(ctx.currentBatch);

Review comment (performance):
We don't need to mock timings or metrics in the new tests. We can use `testScheduleWriteOperationWithBatching` as a template for the new tests.
```suggestion
        MockTimer timer = new MockTimer();
        MockPartitionWriter writer = new MockPartitionWriter();
```

Code after:
            new MockCoordinatorShard.RecordAndMetadata(3, records.get(3))
        ), ctx.coordinator.coordinator().fullRecords());
        assertEquals(List.of(
            records(secondBatchTimestamp, records.subList(0, 3)),
            records(secondBatchTimestamp, records.subList(3, 4))
        ), writer.entries(TP));
        verify(runtimeMetrics, times(2)).recordFlushTime(10);

        // Commit and verify that writes are completed.
        writer.commit(TP);
        assertTrue(write1.isDone());
        assertTrue(write2.isDone());
        assertTrue(write3.isDone());
        assertEquals(4L, ctx.coordinator.lastCommittedOffset());
        assertEquals("response1", write1.get(5, TimeUnit.SECONDS));
        assertEquals("response2", write2.get(5, TimeUnit.SECONDS));
        assertEquals("response3", write3.get(5, TimeUnit.SECONDS));
    }

    @Test
    public void testCompressibleRecordTriggersFlushAndSucceeds() throws Exception {
        MockTimer timer = new MockTimer();
        MockPartitionWriter writer = new MockPartitionWriter();
        Compression compression = Compression.gzip().build();

        CoordinatorRuntime<MockCoordinatorShard, String> runtime =
            new CoordinatorRuntime.Builder<MockCoordinatorShard, String>()
                .withTime(timer.time())
                .withTimer(timer)
                .withDefaultWriteTimeOut(Duration.ofMillis(20))
                .withLoader(new MockCoordinatorLoader())
                .withEventProcessor(new DirectEventProcessor())
                .withPartitionWriter(writer)
                .withCoordinatorShardBuilderSupplier(new MockCoordinatorShardBuilderSupplier())
                .withCoordinatorRuntimeMetrics(mock(CoordinatorRuntimeMetrics.class))
                .withCoordinatorMetrics(mock(CoordinatorMetrics.class))
                .withCompression(compression)
                .withSerializer(new StringSerializer())
                .withAppendLingerMs(10)
                .withExecutorService(mock(ExecutorService.class))
                .build();

        // Schedule the loading.
        runtime.scheduleLoadOperation(TP, 10);

        // Verify the initial state.
        CoordinatorRuntime<MockCoordinatorShard, String>.CoordinatorContext ctx = runtime.contextOrThrow(TP);
        assertNull(ctx.currentBatch);

        // Get the max batch size.
        int maxBatchSize = writer.config(TP).maxMessageSize();