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

        try {
            validateOffsetFetch(request, lastCommittedOffset);
        } catch (GroupIdNotFoundException ex) {
            return new OffsetFetchResponseData.OffsetFetchResponseGroup()
                .setGroupId(request.groupId())
                .setTopics(Collections.emptyList());
        }

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

        if (groupOffsets != null) {
            groupOffsets.entrySet(lastCommittedOffset).forEach(topicEntry -> {
                final String topic = topicEntry.getKey();
                final TimelineHashMap<Integer, OffsetAndMetadata> topicOffsets = topicEntry.getValue();

                final OffsetFetchResponseData.OffsetFetchResponseTopics topicResponse =
                    new OffsetFetchResponseData.OffsetFetchResponseTopics().setName(topic);
                topicResponses.add(topicResponse);

                topicOffsets.entrySet(lastCommittedOffset).forEach(partitionEntry -> {
                    final int partition = partitionEntry.getKey();
                    final OffsetAndMetadata offsetAndMetadata = partitionEntry.getValue();

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

        return new OffsetFetchResponseData.OffsetFetchResponseGroup()
            .setGroupId(request.groupId())
            .setTopics(topicResponses);
    }
@Test
    public void testFetchAllOffsetsWithUnknownGroup() {
        OffsetMetadataManagerTestContext context = new OffsetMetadataManagerTestContext.Builder().build();
        assertEquals(Collections.emptyList(), context.fetchAllOffsets("group", Long.MAX_VALUE));
    }