smjn commented on code in PR #17011: URL: https://github.com/apache/kafka/pull/17011#discussion_r1736540128
########## share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java: ########## @@ -0,0 +1,523 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.coordinator.share; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.compress.Compression; +import org.apache.kafka.common.config.TopicConfig; +import org.apache.kafka.common.internals.Topic; +import org.apache.kafka.common.message.ReadShareGroupStateRequestData; +import org.apache.kafka.common.message.ReadShareGroupStateResponseData; +import org.apache.kafka.common.message.WriteShareGroupStateRequestData; +import org.apache.kafka.common.message.WriteShareGroupStateResponseData; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.requests.ReadShareGroupStateResponse; +import org.apache.kafka.common.requests.RequestContext; +import org.apache.kafka.common.requests.WriteShareGroupStateResponse; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.coordinator.common.runtime.CoordinatorEventProcessor; +import org.apache.kafka.coordinator.common.runtime.CoordinatorLoader; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRecord; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntime; +import org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetrics; +import org.apache.kafka.coordinator.common.runtime.CoordinatorShardBuilderSupplier; +import org.apache.kafka.coordinator.common.runtime.MultiThreadedEventProcessor; +import org.apache.kafka.coordinator.common.runtime.PartitionWriter; +import org.apache.kafka.coordinator.share.metrics.ShareCoordinatorMetrics; +import org.apache.kafka.server.config.ShareCoordinatorConfig; +import org.apache.kafka.server.group.share.SharePartitionKey; +import org.apache.kafka.server.record.BrokerCompressionType; +import org.apache.kafka.server.util.timer.Timer; + +import org.slf4j.Logger; + +import java.time.Duration; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.OptionalInt; +import java.util.Properties; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.IntSupplier; +import java.util.stream.Collectors; + +public class ShareCoordinatorService implements ShareCoordinator { + private final ShareCoordinatorConfig config; + private final Logger log; + private final AtomicBoolean isActive = new AtomicBoolean(false); // for controlling start and stop + private final CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime; + private final ShareCoordinatorMetrics shareCoordinatorMetrics; + private volatile int numPartitions = -1; // Number of partitions for __share_group_state. Provided when component is started. + private final Time time; + + public static class Builder { + private final int nodeId; + private final ShareCoordinatorConfig config; + private PartitionWriter writer; + private CoordinatorLoader<CoordinatorRecord> loader; + private Time time; + private Timer timer; + + private ShareCoordinatorMetrics coordinatorMetrics; + private CoordinatorRuntimeMetrics coordinatorRuntimeMetrics; + + public Builder(int nodeId, ShareCoordinatorConfig config) { + this.nodeId = nodeId; + this.config = config; + } + + public Builder withWriter(PartitionWriter writer) { + this.writer = writer; + return this; + } + + public Builder withLoader(CoordinatorLoader<CoordinatorRecord> loader) { + this.loader = loader; + return this; + } + + public Builder withTime(Time time) { + this.time = time; + return this; + } + + public Builder withTimer(Timer timer) { + this.timer = timer; + return this; + } + + public Builder withCoordinatorMetrics(ShareCoordinatorMetrics coordinatorMetrics) { + this.coordinatorMetrics = coordinatorMetrics; + return this; + } + + public Builder withCoordinatorRuntimeMetrics(CoordinatorRuntimeMetrics coordinatorRuntimeMetrics) { + this.coordinatorRuntimeMetrics = coordinatorRuntimeMetrics; + return this; + } + + public ShareCoordinatorService build() { + if (config == null) { + throw new IllegalArgumentException("Config must be set."); + } + if (writer == null) { + throw new IllegalArgumentException("Writer must be set."); + } + if (loader == null) { + throw new IllegalArgumentException("Loader must be set."); + } + if (time == null) { + throw new IllegalArgumentException("Time must be set."); + } + if (timer == null) { + throw new IllegalArgumentException("Timer must be set."); + } + if (coordinatorMetrics == null) { + throw new IllegalArgumentException("Share Coordinator metrics must be set."); + } + if (coordinatorRuntimeMetrics == null) { + throw new IllegalArgumentException("Coordinator runtime metrics must be set."); + } + + String logPrefix = String.format("ShareCoordinator id=%d", nodeId); + LogContext logContext = new LogContext(String.format("[%s] ", logPrefix)); + + CoordinatorShardBuilderSupplier<ShareCoordinatorShard, CoordinatorRecord> supplier = () -> + new ShareCoordinatorShard.Builder(config); + + CoordinatorEventProcessor processor = new MultiThreadedEventProcessor( + logContext, + "share-coordinator-event-processor-", + config.shareCoordinatorNumThreads(), + time, + coordinatorRuntimeMetrics + ); + + CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime = + new CoordinatorRuntime.Builder<ShareCoordinatorShard, CoordinatorRecord>() + .withTime(time) + .withTimer(timer) + .withLogPrefix(logPrefix) + .withLogContext(logContext) + .withEventProcessor(processor) + .withPartitionWriter(writer) + .withLoader(loader) + .withCoordinatorShardBuilderSupplier(supplier) + .withTime(time) + .withDefaultWriteTimeOut(Duration.ofMillis(config.shareCoordinatorWriteTimeoutMs())) + .withCoordinatorRuntimeMetrics(coordinatorRuntimeMetrics) + .withCoordinatorMetrics(coordinatorMetrics) + .withSerializer(new ShareCoordinatorRecordSerde()) + .withCompression(Compression.of(config.shareCoordinatorStateTopicCompressionType()).build()) + .withAppendLingerMs(config.shareCoordinatorAppendLingerMs()) + .build(); + + return new ShareCoordinatorService( + logContext, + config, + runtime, + coordinatorMetrics, + time + ); + } + } + + public ShareCoordinatorService( + LogContext logContext, + ShareCoordinatorConfig config, + CoordinatorRuntime<ShareCoordinatorShard, CoordinatorRecord> runtime, + ShareCoordinatorMetrics shareCoordinatorMetrics, + Time time) { + this.log = logContext.logger(ShareCoordinatorService.class); + this.config = config; + this.runtime = runtime; + this.shareCoordinatorMetrics = shareCoordinatorMetrics; + this.time = time; + } + + @Override + public int partitionFor(String key) { + return Utils.abs(key.hashCode()) % numPartitions; + } + + @Override + public Properties shareGroupStateTopicConfigs() { + Properties properties = new Properties(); + properties.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE); // as defined in KIP-932 + properties.put(TopicConfig.COMPRESSION_TYPE_CONFIG, BrokerCompressionType.PRODUCER.name); + properties.put(TopicConfig.SEGMENT_BYTES_CONFIG, config.shareCoordinatorStateTopicSegmentBytes()); + return properties; + } + + @Override + public void startup( Review Comment: ``` The share coordinator startup method will get initialized from BrokerMetadataPublisher. At the time of writing, the publisher uses metadata cache to fetch the number of partitions of the share state topic. In case this information is not available, the user provided config will be used to fetch the value. This is consistent with the group coordinator startup functionality. ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
