@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();
    }
@Test
    public void testOnPartitionsDeletedWhenServiceIsNotStarted() {
        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
        GroupCoordinatorService service = new GroupCoordinatorService(
            new LogContext(),
            createConfig(),
            runtime,
            new GroupCoordinatorMetrics(),
            createConfigManager()
        );

        assertThrows(CoordinatorNotAvailableException.class, () -> service.onPartitionsDeleted(
            Collections.singletonList(new TopicPartition("foo", 0)),
            BufferSupplier.NO_CACHING
        ));
    }