@Test
    public void testUseDefaultLeaderEpochV8() {
        final Optional<Integer> emptyLeaderEpoch = Optional.empty();
        partitionDataMap.clear();

        partitionDataMap.put(new TopicPartition(topicOne, partitionOne),
            new PartitionData(
                offset,
                emptyLeaderEpoch,
                metadata,
                Errors.UNKNOWN_TOPIC_OR_PARTITION)
        );
        OffsetFetchResponse response = new OffsetFetchResponse(
            throttleTimeMs,
            Collections.singletonMap(groupOne, Errors.NOT_COORDINATOR),
            Collections.singletonMap(groupOne, partitionDataMap));
        OffsetFetchResponseData expectedData =
            new OffsetFetchResponseData()
                .setGroups(Collections.singletonList(
                    new OffsetFetchResponseGroup()
                        .setGroupId(groupOne)
                        .setTopics(Collections.singletonList(
                            new OffsetFetchResponseTopics()
                                .setName(topicOne)
                                .setPartitions(Collections.singletonList(
                                    new OffsetFetchResponsePartitions()
                                        .setPartitionIndex(partitionOne)
                                        .setCommittedOffset(offset)
                                        .setCommittedLeaderEpoch(RecordBatch.NO_PARTITION_LEADER_EPOCH)
                                        .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code())
                                        .setMetadata(metadata)))))
                        .setErrorCode(Errors.NOT_COORDINATOR.code())))
                .setThrottleTimeMs(throttleTimeMs);
        assertEquals(expectedData, response.data());
    }
Test behavior changes over the versions. Refer to resources.common.messages.OffsetFetchResponse.json
clients/src/test/java/org/apache/kafka/common/requests/OffsetFetchResponseTest.java