@Test
    public void testIncrementals() {
        Map<String, Uuid> topicIds = new HashMap<>();
        Map<Uuid, String> topicNames = new HashMap<>();
        // We want to test both on older versions that do not use topic IDs and on newer versions that do.
        List<Short> versions = Arrays.asList((short) 12, ApiKeys.FETCH.latestVersion());
        versions.forEach(version -> {
            FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
            FetchSessionHandler.Builder builder = handler.newBuilder();
            addTopicId(topicIds, topicNames, "foo", version);
            Uuid fooId = topicIds.getOrDefault("foo", Uuid.ZERO_UUID);
            TopicPartition foo0 = new TopicPartition("foo", 0);
            TopicPartition foo1 = new TopicPartition("foo", 1);
            builder.add(foo0, new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
            builder.add(foo1, new FetchRequest.PartitionData(fooId, 10, 110, 210, Optional.empty()));
            FetchSessionHandler.FetchRequestData data = builder.build();
            assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200),
                    new ReqEntry("foo", fooId, 1, 10, 110, 210)),
                    data.toSend(), data.sessionPartitions());
            assertEquals(INVALID_SESSION_ID, data.metadata().sessionId());
            assertEquals(INITIAL_EPOCH, data.metadata().epoch());

            FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123,
                respMap(new RespEntry("foo", 0, fooId, 10, 20),
                        new RespEntry("foo", 1, fooId, 10, 20)));
            handler.handleResponse(resp, version);

            // Test an incremental fetch request which adds one partition and modifies another.
            FetchSessionHandler.Builder builder2 = handler.newBuilder();
            addTopicId(topicIds, topicNames, "bar", version);
            Uuid barId = topicIds.getOrDefault("bar", Uuid.ZERO_UUID);
            TopicPartition bar0 = new TopicPartition("bar", 0);
            builder2.add(foo0, new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
            builder2.add(foo1, new FetchRequest.PartitionData(fooId, 10, 120, 210, Optional.empty()));
            builder2.add(bar0, new FetchRequest.PartitionData(barId, 20, 200, 200, Optional.empty()));
            FetchSessionHandler.FetchRequestData data2 = builder2.build();
            assertFalse(data2.metadata().isFull());
            assertMapEquals(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200),
                    new ReqEntry("foo", fooId, 1, 10, 120, 210),
                    new ReqEntry("bar", barId, 0, 20, 200, 200)),
                    data2.sessionPartitions());
            assertMapEquals(reqMap(new ReqEntry("bar", barId, 0, 20, 200, 200),
                    new ReqEntry("foo", fooId, 1, 10, 120, 210)),
                    data2.toSend());

            FetchResponse resp2 = FetchResponse.of(Errors.NONE, 0, 123,
                respMap(new RespEntry("foo", 1, fooId, 20, 20)));
            handler.handleResponse(resp2, version);

            // Skip building a new request.  Test that handling an invalid fetch session epoch response results
            // in a request which closes the session.
            FetchResponse resp3 = FetchResponse.of(Errors.INVALID_FETCH_SESSION_EPOCH, 0, INVALID_SESSION_ID,
                respMap());
            handler.handleResponse(resp3, version);

            FetchSessionHandler.Builder builder4 = handler.newBuilder();
            builder4.add(foo0, new FetchRequest.PartitionData(fooId, 0, 100, 200, Optional.empty()));
            builder4.add(foo1, new FetchRequest.PartitionData(fooId, 10, 120, 210, Optional.empty()));
            builder4.add(bar0, new FetchRequest.PartitionData(barId, 20, 200, 200, Optional.empty()));
            FetchSessionHandler.FetchRequestData data4 = builder4.build();
            assertTrue(data4.metadata().isFull());
            assertEquals(data2.metadata().sessionId(), data4.metadata().sessionId());
            assertEquals(INITIAL_EPOCH, data4.metadata().epoch());
            assertMapsEqual(reqMap(new ReqEntry("foo", fooId, 0, 0, 100, 200),
                    new ReqEntry("foo", fooId, 1, 10, 120, 210),
                    new ReqEntry("bar", barId, 0, 20, 200, 200)),
                    data4.sessionPartitions(), data4.toSend());
        });
    }
Test handling an incremental fetch session.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java