Repository: apache/kafka
File: share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java (Java)

Code before:
        );
    }

    @Override
    public void onResignation(int partitionIndex, OptionalInt partitionLeaderEpoch) {
        throwIfNotActive();
        TopicPartition tp = new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, partitionIndex);
        lastPrunedOffsets.remove(tp);
        runtime.scheduleUnloadOperation(
            tp,
            partitionLeaderEpoch
        );
    }

    @Override
    public void onTopicsDeleted(Set<Uuid> deletedTopicIds, BufferSupplier bufferSupplier) throws ExecutionException, InterruptedException {
        throwIfNotActive();
        if (deletedTopicIds.isEmpty()) {
            return;
        }
        CompletableFuture.allOf(
            FutureUtils.mapExceptionally(
                runtime.scheduleWriteAllOperation(
                    "on-topics-deleted",
                    coordinator -> coordinator.maybeCleanupShareState(deletedTopicIds)
                ),
                exception -> {
                    log.error("Received error while trying to cleanup deleted topics.", exception);
                    return null;
                }
            ).toArray(new CompletableFuture<?>[0])
        ).get();
    }

    @Override
    public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage) {
        throwIfNotActive();
        this.runtime.onMetadataUpdate(
            new KRaftCoordinatorMetadataDelta(Objects.requireNonNull(delta, "delta must be provided")),
            new KRaftCoordinatorMetadataImage(Objects.requireNonNull(newImage, "newImage must be provided"))
        );
        boolean enabled = isShareGroupsEnabled(newImage);
        // enabled    shouldRunJob         result (XOR)
        // 0            0               no op on flag, do not call jobs
        // 0            1               disable flag, do not call jobs                      => action
        // 1            0               enable flag, call jobs as they are not recursing    => action
        // 1            1               no op on flag, do not call jobs
        if (enabled ^ shouldRunPeriodicJob) {
            shouldRunPeriodicJob = enabled;
            if (enabled) {
                setupPeriodicJobs();

Review comment (suggestion):
Previously, `handleTopicsDeletion` executed before `onMetadataUpdate`, but now the order is reversed. I think the current order makes more sense

Code after:

    @Override
    public void onResignation(int partitionIndex, OptionalInt partitionLeaderEpoch) {
        throwIfNotActive();
        TopicPartition tp = new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, partitionIndex);
        lastPrunedOffsets.remove(tp);
        runtime.scheduleUnloadOperation(
            tp,
            partitionLeaderEpoch
        );
    }

    @Override
    public void onMetadataUpdate(MetadataDelta delta, MetadataImage newImage) {
        throwIfNotActive();
        Objects.requireNonNull(delta, "delta must be provided");
        Objects.requireNonNull(newImage, "newImage must be provided");

        this.runtime.onMetadataUpdate(
            new KRaftCoordinatorMetadataDelta(delta),
            new KRaftCoordinatorMetadataImage(newImage)
        );

        // Handle topic deletions from the delta.
        if (delta.topicsDelta() != null && !delta.topicsDelta().deletedTopicIds().isEmpty()) {
            handleTopicsDeletion(delta.topicsDelta().deletedTopicIds());
        }

        boolean enabled = isShareGroupsEnabled(newImage);
        // enabled    shouldRunJob         result (XOR)
        // 0            0               no op on flag, do not call jobs
        // 0            1               disable flag, do not call jobs                      => action
        // 1            0               enable flag, call jobs as they are not recursing    => action
        // 1            1               no op on flag, do not call jobs
        if (enabled ^ shouldRunPeriodicJob) {
            shouldRunPeriodicJob = enabled;
            if (enabled) {
                setupPeriodicJobs();
            }
        }
    }

    private void handleTopicsDeletion(Set<Uuid> deletedTopicIds) {
        CompletableFuture.allOf(
            FutureUtils.mapExceptionally(
                runtime.scheduleWriteAllOperation(
                    "on-topics-deleted",
                    coordinator -> coordinator.maybeCleanupShareState(deletedTopicIds)
                ),
                exception -> {
                    log.error("Received error while trying to cleanup deleted topics.", exception);