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

Code before:
        throwIfNotActive();
        if (deletedTopicIds.isEmpty()) {
            return;
        }
        CompletableFuture.allOf(
            FutureUtils.mapExceptionally(
                runtime.scheduleWriteAllOperation(
                    "on-topics-deleted",
                    Duration.ofMillis(config.shareCoordinatorWriteTimeoutMs()),
                    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 onNewMetadataImage(CoordinatorMetadataImage newImage, FeaturesImage newFeaturesImage, CoordinatorMetadataDelta delta) {
        throwIfNotActive();
        this.runtime.onMetadataUpdate(delta, newImage);
        boolean enabled = isShareGroupsEnabled(newFeaturesImage);
        // 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();
            }
        }
    }

    TopicPartition topicPartitionFor(SharePartitionKey key) {
        return new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, partitionFor(key));
    }

    private static <P> boolean isEmpty(List<P> list) {
        return list == null || list.isEmpty();
    }

    private void throwIfNotActive() {
        if (!isActive.get()) {
            throw Errors.COORDINATOR_NOT_AVAILABLE.exception();
        }
    }


Review comment (refactor):
some methods of `KRaftCoordinatorMetadataDelta` assume `metadataDelta` can be null, but other methods, such as `hashCode` and `equals`, don't handle `null` value. I guess you will have a follow-up patch to refactor all the null-handling logic at once

Code after:
    @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",
                    Duration.ofMillis(config.shareCoordinatorWriteTimeoutMs()),
                    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();
            }
        }
    }

    TopicPartition topicPartitionFor(SharePartitionKey key) {
        return new TopicPartition(Topic.SHARE_GROUP_STATE_TOPIC_NAME, partitionFor(key));
    }

    private static <P> boolean isEmpty(List<P> list) {
        return list == null || list.isEmpty();
    }

    private void throwIfNotActive() {