@Override
    public void onPartitionsDeleted(
        List<TopicPartition> topicPartitions,
        BufferSupplier bufferSupplier
    ) throws ExecutionException, InterruptedException {
        throwIfNotActive();

        CompletableFuture.allOf(
            FutureUtils.mapExceptionally(
                runtime.scheduleWriteAllOperation(
                    "on-partition-deleted",
                    Duration.ofMillis(config.offsetCommitTimeoutMs()),
                    coordinator -> coordinator.onPartitionsDeleted(topicPartitions)
                ),
                exception -> {
                    log.error("Could not delete offsets for deleted partitions {} due to: {}.",
                        topicPartitions, exception.getMessage(), exception
                    );
                    return null;
                }
            ).toArray(new CompletableFuture[0])
        ).get();
    }
See {@link GroupCoordinator#onPartitionsDeleted(List, BufferSupplier)}.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java