@ParameterizedTest
    @MethodSource("idUsageCombinations")
    public void testTopicIdReplaced(boolean startsWithTopicIds, boolean endsWithTopicIds) {
        TopicPartition tp = new TopicPartition("foo", 0);
        FetchSessionHandler handler = new FetchSessionHandler(LOG_CONTEXT, 1);
        FetchSessionHandler.Builder builder = handler.newBuilder();
        Uuid topicId1 = startsWithTopicIds ? Uuid.randomUuid() : Uuid.ZERO_UUID;
        builder.add(tp, new FetchRequest.PartitionData(topicId1, 0, 100, 200, Optional.empty()));
        FetchSessionHandler.FetchRequestData data = builder.build();
        assertMapsEqual(reqMap(new ReqEntry("foo", topicId1, 0, 0, 100, 200)),
                data.toSend(), data.sessionPartitions());
        assertTrue(data.metadata().isFull());
        assertEquals(startsWithTopicIds, data.canUseTopicIds());

        FetchResponse resp = FetchResponse.of(Errors.NONE, 0, 123, respMap(new RespEntry("foo", 0, topicId1, 10, 20)));
        short version = startsWithTopicIds ? ApiKeys.FETCH.latestVersion() : 12;
        handler.handleResponse(resp, version);

        // Try to add a new topic ID.
        FetchSessionHandler.Builder builder2 = handler.newBuilder();
        Uuid topicId2 = endsWithTopicIds ? Uuid.randomUuid() : Uuid.ZERO_UUID;
        // Use the same data besides the topic ID.
        FetchRequest.PartitionData partitionData = new FetchRequest.PartitionData(topicId2, 0, 100, 200, Optional.empty());
        builder2.add(tp, partitionData);
        FetchSessionHandler.FetchRequestData data2 = builder2.build();

        if (startsWithTopicIds && endsWithTopicIds) {
            // If we started with an ID, both a only a new ID will count towards replaced.
            // The old topic ID partition should be in toReplace, and the new one should be in toSend.
            assertEquals(Collections.singletonList(new TopicIdPartition(topicId1, tp)), data2.toReplace());
            assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)),
                    data2.toSend(), data2.sessionPartitions());

            // sessionTopicNames should contain only the second topic ID.
            assertEquals(Collections.singletonMap(topicId2, tp.topic()), handler.sessionTopicNames());

        } else if (startsWithTopicIds || endsWithTopicIds) {
            // If we downgraded to not using topic IDs we will want to send this data.
            // However, we will not mark the partition as one replaced. In this scenario, we should see the session close due to
            // changing request types.
            // We will have the new topic ID in the session partition map
            assertEquals(Collections.emptyList(), data2.toReplace());
            assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)),
                    data2.toSend(), data2.sessionPartitions());
            // The topicNames map will have the new topic ID if it is valid.
            // The old topic ID should be removed as the map will be empty if the request doesn't use topic IDs.
            if (endsWithTopicIds) {
                assertEquals(Collections.singletonMap(topicId2, tp.topic()), handler.sessionTopicNames());
            } else {
                assertEquals(Collections.emptyMap(), handler.sessionTopicNames());
            }

        } else {
            // Otherwise, we have no partition in toReplace and since the partition and topic ID was not updated, there is no data to send.
            assertEquals(Collections.emptyList(), data2.toReplace());
            assertEquals(Collections.emptyMap(), data2.toSend());
            assertMapsEqual(reqMap(new ReqEntry("foo", topicId2, 0, 0, 100, 200)), data2.sessionPartitions());
            // There is also nothing in the sessionTopicNames map, as there are no topic IDs used.
            assertEquals(Collections.emptyMap(), handler.sessionTopicNames());
        }

        // Should have the same session ID, and next epoch and can use topic IDs if it ended with topic IDs.
        assertEquals(123, data2.metadata().sessionId(), "Did not use same session");
        assertEquals(1, data2.metadata().epoch(), "Did not have correct epoch");
        assertEquals(endsWithTopicIds, data2.canUseTopicIds());
    }
Test that calling FetchSessionHandler#Builder#build twice fails.
clients/src/test/java/org/apache/kafka/clients/FetchSessionHandlerTest.java