KAFKA-14462; [17/N] Add CoordinatorRuntime

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/Coordinator.java

@@ -0,0 +1,36 @@
+/*
+ * 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.group.runtime;
+
+/**
+ * Coordinator is basically a replicated state machine managed by the
+ * {@link CoordinatorRuntime}.
+ */
+public interface Coordinator<U> extends CoordinatorPlayback<U> {
+
+    /**
+     * The coordinator has been loaded. This is used to apply any
+     * post loading operations (e.g. registering timers).
+     */
+    default void onLoaded() {};
+
+    /**
+     * The coordinator has been unloaded. This is used to apply
+     * any post unloading operations.
+     */
+    default void onUnloaded() {};
+}

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilder.java

@@ -0,0 +1,45 @@
+/*
+ * 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.group.runtime;
+
+import org.apache.kafka.timeline.SnapshotRegistry;
+
+/**
+ * A builder to build a {@link Coordinator} replicated state machine.
+ *
+ * @param <S> The type of the coordinator.
+ * @param <U> The record type.
+ */
+interface CoordinatorBuilder<S extends Coordinator<U>, U> {
+
+    /**
+     * Sets the snapshot registry used to back all the timeline
+     * datastructures used by the coordinator.
+     *
+     * @param snapshotRegistry The registry.
+     *
+     * @return The builder.
+     */
+    CoordinatorBuilder<S, U> withSnapshotRegistry(
+        SnapshotRegistry snapshotRegistry
+    );
+
+    /**
+     * @return The built coordinator.
+     */
+    S build();
+}

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilderSupplier.java

@@ -0,0 +1,30 @@
+/*
+ * 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.group.runtime;
+
+/**
+ * Supplies a {@link CoordinatorBuilder} to the {@link CoordinatorRuntime}.
+ *
+ * @param <S> The type of the coordinator.
+ * @param <U> The record type.
+ */
+interface CoordinatorBuilderSupplier<S extends Coordinator<U>, U> {
+    /**
+     * @return A {@link CoordinatorBuilder}.
+     */
+    CoordinatorBuilder<S, U> get();
+}

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorEvent.java

@@ -16,21 +16,23 @@
  */
 package org.apache.kafka.coordinator.group.runtime;
 
+import org.apache.kafka.common.TopicPartition;
+
 /**
  * The base event type used by all events processed in the
  * coordinator runtime.
  */
-public interface CoordinatorEvent extends EventAccumulator.Event<Integer> {
+public interface CoordinatorEvent extends EventAccumulator.Event<TopicPartition> {
 
     /**
-     * Runs the event.
+     * Executes the event.
      */
     void run();
 
     /**
      * Completes the event with the provided exception.
      *
-     * @param exception An exception to complete the event with.
+     * @param exception An exception if the processing of the event failed or null otherwise.
      */
     void complete(Throwable exception);
 }