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

Code before:
        );

        // Verify the state.
        assertEquals(3L, ctx.coordinator.lastWrittenOffset());
        assertEquals(0L, ctx.coordinator.lastCommittedOffset());
        assertEquals(List.of(
            new MockCoordinatorShard.RecordAndMetadata(0, records.get(0)),
            new MockCoordinatorShard.RecordAndMetadata(1, records.get(1)),
            new MockCoordinatorShard.RecordAndMetadata(2, largeRecord.get(0))
        ), ctx.coordinator.coordinator().fullRecords());
        assertEquals(List.of(
            records(firstBatchTimestamp, compression, records),
            records(secondBatchTimestamp, compression, largeRecord)
        ), writer.entries(TP));

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

    @Test
    public void testLargeUncompressibleRecordTriggersFlushAndFails() 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.

Review comment (performance):
Could we add a variant of this test that would try to pack the large record into the same batch as write1 under the old implementation? 3x of the max batch size is too large for that and we'd need something like 0.75 of the max batch size. We can use `@ParameterizedTest`, `@ValueSource` and add a double parameter for the fraction of the max batch size to use.

Code after:
        // Verify the state. The first batch has flushed but the second is pending.
        assertEquals(2L, ctx.coordinator.lastWrittenOffset());
        assertEquals(0L, ctx.coordinator.lastCommittedOffset());
        assertEquals(List.of(
            new MockCoordinatorShard.RecordAndMetadata(0, records.get(0)),
            new MockCoordinatorShard.RecordAndMetadata(1, records.get(1)),
            new MockCoordinatorShard.RecordAndMetadata(2, largeRecord.get(0))
        ), ctx.coordinator.coordinator().fullRecords());
        assertEquals(List.of(
            records(firstBatchTimestamp, compression, records)
        ), writer.entries(TP));

        // Advance past the linger time
        timer.advanceClock(11);

        // Commit and verify that the second batch is completed
        writer.commit(TP);
        assertTrue(write1.isDone());
        assertTrue(write2.isDone());
        assertEquals(3L, ctx.coordinator.lastCommittedOffset());
        assertEquals("response1", write1.get(5, TimeUnit.SECONDS));
        assertEquals("response2", write2.get(5, TimeUnit.SECONDS));
    }

    @Test
    public void testLargeCompressibleRecordTriggersFlushAndSucceeds() 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.