public OffsetFetchResponseData.OffsetFetchResponseGroup fetchOffsets(
        OffsetFetchRequestData.OffsetFetchRequestGroup request,
        long lastCommittedOffset
    ) throws ApiException {
        final boolean requireStable = lastCommittedOffset == Long.MAX_VALUE;

        boolean failAllPartitions = false;
        try {
            validateOffsetFetch(request, lastCommittedOffset);
        } catch (GroupIdNotFoundException ex) {
            failAllPartitions = true;
        }

        final List<OffsetFetchResponseData.OffsetFetchResponseTopics> topicResponses = new ArrayList<>(request.topics().size());
        final TimelineHashMap<String, TimelineHashMap<Integer, OffsetAndMetadata>> groupOffsets =
            failAllPartitions ? null : offsets.offsetsByGroup.get(request.groupId(), lastCommittedOffset);

        request.topics().forEach(topic -> {
            final OffsetFetchResponseData.OffsetFetchResponseTopics topicResponse =
                new OffsetFetchResponseData.OffsetFetchResponseTopics().setName(topic.name());
            topicResponses.add(topicResponse);

            final TimelineHashMap<Integer, OffsetAndMetadata> topicOffsets = groupOffsets == null ?
                null : groupOffsets.get(topic.name(), lastCommittedOffset);

            topic.partitionIndexes().forEach(partitionIndex -> {
                final OffsetAndMetadata offsetAndMetadata = topicOffsets == null ?
                    null : topicOffsets.get(partitionIndex, lastCommittedOffset);

                if (requireStable && hasPendingTransactionalOffsets(request.groupId(), topic.name(), partitionIndex)) {
                    topicResponse.partitions().add(new OffsetFetchResponseData.OffsetFetchResponsePartitions()
                        .setPartitionIndex(partitionIndex)
                        .setErrorCode(Errors.UNSTABLE_OFFSET_COMMIT.code())
                        .setCommittedOffset(INVALID_OFFSET)
                        .setCommittedLeaderEpoch(-1)
                        .setMetadata(""));
                } else if (offsetAndMetadata == null) {
                    topicResponse.partitions().add(new OffsetFetchResponseData.OffsetFetchResponsePartitions()
                        .setPartitionIndex(partitionIndex)
                        .setCommittedOffset(INVALID_OFFSET)
                        .setCommittedLeaderEpoch(-1)
                        .setMetadata(""));
                } else {
                    topicResponse.partitions().add(new OffsetFetchResponseData.OffsetFetchResponsePartitions()
                        .setPartitionIndex(partitionIndex)
                        .setCommittedOffset(offsetAndMetadata.committedOffset)
                        .setCommittedLeaderEpoch(offsetAndMetadata.leaderEpoch.orElse(-1))
                        .setMetadata(offsetAndMetadata.metadata));
                }
            });
        });

        return new OffsetFetchResponseData.OffsetFetchResponseGroup()
            .setGroupId(request.groupId())
            .setTopics(topicResponses);
    }
@Test
    public void testFetchOffsetsWithUnknownGroup() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();

        List<OffsetFetchRequestData.OffsetFetchRequestTopics> request = Arrays.asList(
            new OffsetFetchRequestData.OffsetFetchRequestTopics()
                .setName("foo")
                .setPartitionIndexes(Arrays.asList(0, 1)),
            new OffsetFetchRequestData.OffsetFetchRequestTopics()
                .setName("bar")
                .setPartitionIndexes(Collections.singletonList(0))
        );

        List<OffsetFetchResponseData.OffsetFetchResponseTopics> expectedResponse = Arrays.asList(
            new OffsetFetchResponseData.OffsetFetchResponseTopics()
                .setName("foo")
                .setPartitions(Arrays.asList(
                    mkInvalidOffsetPartitionResponse(0),
                    mkInvalidOffsetPartitionResponse(1)
                )),
            new OffsetFetchResponseData.OffsetFetchResponseTopics()
                .setName("bar")
                .setPartitions(Collections.singletonList(
                    mkInvalidOffsetPartitionResponse(0)
                ))
        );

        assertEquals(expectedResponse, context.fetchOffsets("group", request, Long.MAX_VALUE));
    }