diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/Coordinator.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/Coordinator.java new file mode 100644 index 0000000000000..0eb2499520a2c --- /dev/null +++ b/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 extends CoordinatorPlayback { + + /** + * 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() {}; +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilder.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilder.java new file mode 100644 index 0000000000000..4d9febc14b7d3 --- /dev/null +++ b/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 The type of the coordinator. + * @param The record type. + */ +interface CoordinatorBuilder, U> { + + /** + * Sets the snapshot registry used to back all the timeline + * datastructures used by the coordinator. + * + * @param snapshotRegistry The registry. + * + * @return The builder. + */ + CoordinatorBuilder withSnapshotRegistry( + SnapshotRegistry snapshotRegistry + ); + + /** + * @return The built coordinator. + */ + S build(); +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilderSupplier.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorBuilderSupplier.java new file mode 100644 index 0000000000000..35048177961dc --- /dev/null +++ b/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 The type of the coordinator. + * @param The record type. + */ +interface CoordinatorBuilderSupplier, U> { + /** + * @return A {@link CoordinatorBuilder}. + */ + CoordinatorBuilder get(); +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorEvent.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorEvent.java index 5451cf6c68be1..fb9bdbed6512f 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorEvent.java +++ b/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 { +public interface CoordinatorEvent extends EventAccumulator.Event { /** - * 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); } diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java new file mode 100644 index 0000000000000..4a2c0546f12a2 --- /dev/null +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java @@ -0,0 +1,1081 @@ +/* + * 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.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.CoordinatorLoadInProgressException; +import org.apache.kafka.common.errors.NotCoordinatorException; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.deferred.DeferredEvent; +import org.apache.kafka.deferred.DeferredEventQueue; +import org.apache.kafka.timeline.SnapshotRegistry; +import org.slf4j.Logger; + +import java.util.HashSet; +import java.util.OptionalLong; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * The CoordinatorRuntime provides a framework to implement coordinators such as the group coordinator + * or the transaction coordinator. + * + * The runtime framework maps each underlying partitions (e.g. __consumer_offsets) that that broker is a + * leader of to a coordinator replicated state machine. A replicated state machine holds the hard and soft + * state of all the objects (e.g. groups or offsets) assigned to the partition. The hard state is stored in + * timeline datastructures backed by a SnapshotRegistry. The runtime supports two type of operations + * on state machines: (1) Writes and (2) Reads. + * + * (1) A write operation, aka a request, can read the full and potentially **uncommitted** state from state + * machine to handle the operation. A write operation typically generates a response and a list of + * records. The records are applies to the state machine and persisted to the partition. The response + * is parked until the records are committed and delivered when they are. + * + * (2) A read operation, aka a request, can only read the committed state from the state machine to handle + * the operation. A read operation typically generates a response that is immediately completed. + * + * The runtime framework exposes an asynchronous, future based, API to the world. All the operations + * are executed by an CoordinatorEventProcessor. The processor guarantees that operations for a + * single partition or state machine are not processed concurrently. + * + * @param The type of the state machine. + * @param The type of the record. + */ +public class CoordinatorRuntime, U> { + + /** + * Builder to create a CoordinatorRuntime. + * + * @param The type of the state machine. + * @param The type of the record. + */ + public static class Builder, U> { + private LogContext logContext; + private CoordinatorEventProcessor eventProcessor; + private PartitionWriter partitionWriter; + private CoordinatorLoader loader; + private CoordinatorBuilderSupplier coordinatorBuilderSupplier; + + public Builder withLogContext(LogContext logContext) { + this.logContext = logContext; + return this; + } + + public Builder withEventProcessor(CoordinatorEventProcessor eventProcessor) { + this.eventProcessor = eventProcessor; + return this; + } + + public Builder withPartitionWriter(PartitionWriter partitionWriter) { + this.partitionWriter = partitionWriter; + return this; + } + + public Builder withLoader(CoordinatorLoader loader) { + this.loader = loader; + return this; + } + + public Builder withCoordinatorBuilderSupplier(CoordinatorBuilderSupplier coordinatorBuilderSupplier) { + this.coordinatorBuilderSupplier = coordinatorBuilderSupplier; + return this; + } + + public CoordinatorRuntime build() { + if (logContext == null) + logContext = new LogContext(); + if (eventProcessor == null) + throw new IllegalArgumentException("Event processor must be set."); + if (partitionWriter == null) + throw new IllegalArgumentException("Partition write must be set."); + if (loader == null) + throw new IllegalArgumentException("Loader must be set."); + if (coordinatorBuilderSupplier == null) + throw new IllegalArgumentException("State machine supplier must be set."); + + return new CoordinatorRuntime<>( + logContext, + eventProcessor, + partitionWriter, + loader, + coordinatorBuilderSupplier + ); + } + } + + /** + * The various state that a coordinator for a partition can be in. + */ + enum CoordinatorState { + /** + * Initial state when a coordinator is created. + */ + INITIAL { + @Override + boolean canTransitionFrom(CoordinatorState state) { + return false; + } + }, + + /** + * The coordinator is being loaded. + */ + LOADING { + @Override + boolean canTransitionFrom(CoordinatorState state) { + return state == INITIAL; + } + }, + + /** + * The coordinator is active and can service requests. + */ + ACTIVE { + @Override + boolean canTransitionFrom(CoordinatorState state) { + return state == ACTIVE || state == LOADING; + } + }, + + /** + * The coordinator is closed. + */ + CLOSED { + @Override + boolean canTransitionFrom(CoordinatorState state) { + return true; + } + }, + + /** + * The coordinator loading has failed. + */ + FAILED { + @Override + boolean canTransitionFrom(CoordinatorState state) { + return state == LOADING; + } + }; + + abstract boolean canTransitionFrom(CoordinatorState state); + } + + /** + * CoordinatorContext holds all the metadata around a coordinator state machine. + */ + class CoordinatorContext { + /** + * The topic partition backing the coordinator. + */ + final TopicPartition tp; + + /** + * The snapshot registry backing the coordinator. + */ + final SnapshotRegistry snapshotRegistry; + + /** + * The deferred event queue used to park events waiting + * on records to be committed. + */ + final DeferredEventQueue deferredEventQueue; + + /** + * The current state. + */ + volatile CoordinatorState state; + + /** + * The actual state machine. + */ + volatile S coordinator; + + /** + * The current epoch of the coordinator. This represents + * the epoch of the partition leader. + */ + volatile int epoch; + + /** + * The last offset written to the partition. + */ + volatile long lastWrittenOffset; + + /** + * The last offset committed. This represents the high + * watermark of the partition. + */ + volatile long lastCommittedOffset; + + /** + * Constructor. + * + * @param tp The topic partition of the coordinator. + */ + private CoordinatorContext( + TopicPartition tp + ) { + this.tp = tp; + this.snapshotRegistry = new SnapshotRegistry(logContext); + this.deferredEventQueue = new DeferredEventQueue(logContext); + this.state = CoordinatorState.INITIAL; + this.epoch = -1; + this.lastWrittenOffset = 0L; + this.lastCommittedOffset = 0L; + } + + /** + * Updates the last written offset. This also create a new snapshot + * in the snapshot registry. + * + * @param offset The new last written offset. + */ + private void updateLastWrittenOffset( + long offset + ) { + if (offset <= lastWrittenOffset) { + throw new IllegalStateException("New last written offset " + offset + " of " + tp + + " must be larger than " + lastWrittenOffset + "."); + } + + log.debug("Update last written offset of {} to {}.", tp, offset); + lastWrittenOffset = offset; + snapshotRegistry.getOrCreateSnapshot(offset); + } + + /** + * Reverts the last written offset. This also reverts the snapshot + * registry to this offset. All the changes applied after the offset + * are lost. + * + * @param offset The offset to revert to. + */ + private void revertLastWrittenOffset( + long offset + ) { + if (offset > lastWrittenOffset) { + throw new IllegalStateException("New offset " + offset + " of " + tp + + " must be smaller than " + lastWrittenOffset + "."); + } + + log.debug("Revert last written offset of {} to {}.", tp, offset); + lastWrittenOffset = offset; + snapshotRegistry.revertToSnapshot(offset); + } + + /** + * Updates the last committed offset. This completes all the deferred + * events waiting on this offset. This also cleanups all the snapshots + * prior to this offset. + * + * @param offset The new last committed offset. + */ + private void updateLastCommittedOffset( + long offset + ) { + if (offset <= lastCommittedOffset) { + throw new IllegalStateException("New committed offset " + offset + " of " + tp + + " must be larger than " + lastCommittedOffset + "."); + } + + log.debug("Update committed offset of {} to {}.", tp, offset); + lastCommittedOffset = offset; + deferredEventQueue.completeUpTo(offset); + snapshotRegistry.deleteSnapshotsUpTo(offset); + } + + /** + * Transitions to the new state. + * + * @param newState The new state. + */ + private void transitionTo( + CoordinatorState newState + ) { + if (!newState.canTransitionFrom(state)) { + throw new IllegalStateException("Cannot transition from " + state + " to " + newState); + } + + log.debug("Transition from {} to {}.", state, newState); + switch (newState) { + case LOADING: + state = CoordinatorState.LOADING; + coordinator = coordinatorBuilderSupplier + .get() + .withSnapshotRegistry(snapshotRegistry) + .build(); + break; + + case ACTIVE: + state = CoordinatorState.ACTIVE; + snapshotRegistry.getOrCreateSnapshot(0); + partitionWriter.registerListener(tp, highWatermarklistener); + coordinator.onLoaded(); + break; + + case FAILED: + state = CoordinatorState.FAILED; + unload(); + break; + + case CLOSED: + state = CoordinatorState.CLOSED; + unload(); + break; + + default: + throw new IllegalArgumentException("Transitioning to " + newState + " is not supported."); + } + } + + /** + * Unloads the coordinator. + */ + private void unload() { + partitionWriter.deregisterListener(tp, highWatermarklistener); + deferredEventQueue.failAll(Errors.NOT_COORDINATOR.exception()); + if (coordinator != null) { + coordinator.onUnloaded(); + coordinator = null; + } + } + } + + /** + * A coordinator write operation. + * + * @param The type of the coordinator state machine. + * @param The type of the response. + * @param The type of the records. + */ + public interface CoordinatorWriteOperation { + /** + * Generates the records needed to implement this coordinator write operation. In general, + * this operation should not modify the hard state of the coordinator. That modifications + * will happen later on, when the records generated by this function are applied to the + * coordinator. + * + * @param coordinator The coordinator state machine. + * @return A result containing a list of records and the RPC result. + * @throws KafkaException + */ + CoordinatorResult generateRecordsAndResult(S coordinator) throws KafkaException; + } + + /** + * A coordinator event that modifies the coordinator state. + * + * @param The type of the response. + */ + class CoordinatorWriteEvent implements CoordinatorEvent, DeferredEvent { + /** + * The topic partition that this write event is applied to. + */ + final TopicPartition tp; + + /** + * The operation name. + */ + final String name; + + /** + * The write operation to execute. + */ + final CoordinatorWriteOperation op; + + /** + * The future that will be completed with the response + * generated by the write operation or an error. + */ + final CompletableFuture future; + + /** + * The result of the write operation. It could be null + * if an exception is thrown before it is assigned. + */ + CoordinatorResult result; + + /** + * Constructor. + * + * @param name The operation name. + * @param tp The topic partition that the operation is applied to. + * @param op The write operation. + */ + CoordinatorWriteEvent( + String name, + TopicPartition tp, + CoordinatorWriteOperation op + ) { + this.tp = tp; + this.name = name; + this.op = op; + this.future = new CompletableFuture<>(); + } + + /** + * @return The key used by the CoordinatorEventProcessor to ensure + * that events with the same key are not processed concurrently. + */ + @Override + public TopicPartition key() { + return tp; + } + + /** + * Called by the CoordinatorEventProcessor when the event is executed. + */ + @Override + public void run() { + try { + // Get the context of the coordinator or fail if the coordinator is not in active state. + CoordinatorContext context = activeContextOrThrow(tp); + long prevLastWrittenOffset = context.lastWrittenOffset; + + // Execute the operation. + result = op.generateRecordsAndResult(context.coordinator); + + if (result.records().isEmpty()) { + // If the records are empty, it was a read operation after all. In this case, + // the response can be returned directly iff there are no pending write operations; + // otherwise, the read needs to wait on the last write operation to be completed. + OptionalLong pendingOffset = context.deferredEventQueue.highestPendingOffset(); + if (pendingOffset.isPresent()) { + context.deferredEventQueue.add(pendingOffset.getAsLong(), this); + } else { + complete(null); + } + } else { + // If the records are not empty, first, they are applied to the state machine, + // second, then are written to the partition/log, and finally, the response + // is put into the deferred event queue. + try { + // Apply the records to the state machine. + result.records().forEach(context.coordinator::replay); + + // Write the records to the log and update the last written + // offset. + long offset = partitionWriter.append(tp, result.records()); + context.updateLastWrittenOffset(offset); + + // Add the response to the deferred queue. + if (!future.isDone()) { + context.deferredEventQueue.add(offset, this); + } else { + complete(null); + } + } catch (Throwable t) { + context.revertLastWrittenOffset(prevLastWrittenOffset); + complete(t); + } + } + } catch (Throwable t) { + complete(t); + } + } + + /** + * Completes the future with either the result of the write operation + * or the provided exception. + * + * @param exception The exception to complete the future with. + */ + @Override + public void complete(Throwable exception) { + if (exception == null) { + future.complete(result.response()); + } else { + future.completeExceptionally(exception); + } + } + + @Override + public String toString() { + return "CoordinatorWriteEvent(name=" + name + ")"; + } + } + + /** + * A coordinator read operation. + * + * @param The type of the coordinator state machine. + * @param The type of the response. + */ + public interface CoordinatorReadOperation { + /** + * Generates the response to implement this coordinator read operation. A read + * operation received the last committed offset. It must use it to ensure that + * it does not read uncommitted data from the timeline data structures. + * + * @param state The coordinator state machine. + * @param offset The last committed offset. + * @return A response. + * @throws KafkaException + */ + T generateResponse(S state, long offset) throws KafkaException; + } + + /** + * A coordinator that reads the committed coordinator state. + * + * @param The type of the response. + */ + class CoordinatorReadEvent implements CoordinatorEvent { + /** + * The topic partition that this read event is applied to. + */ + final TopicPartition tp; + + /** + * The operation name. + */ + final String name; + + /** + * The read operation to execute. + */ + final CoordinatorReadOperation op; + + /** + * The future that will be completed with the response + * generated by the read operation or an error. + */ + final CompletableFuture future; + + /** + * The result of the read operation. It could be null + * if an exception is thrown before it is assigned. + */ + T response; + + /** + * Constructor. + * + * @param name The operation name. + * @param tp The topic partition that the operation is applied to. + * @param op The read operation. + */ + CoordinatorReadEvent( + String name, + TopicPartition tp, + CoordinatorReadOperation op + ) { + this.tp = tp; + this.name = name; + this.op = op; + this.future = new CompletableFuture<>(); + } + + /** + * @return The key used by the CoordinatorEventProcessor to ensure + * that events with the same key are not processed concurrently. + */ + @Override + public TopicPartition key() { + return tp; + } + + /** + * Called by the CoordinatorEventProcessor when the event is executed. + */ + @Override + public void run() { + try { + // Get the context of the coordinator or fail if the coordinator is not in active state. + CoordinatorContext context = activeContextOrThrow(tp); + + // Execute the read operation. + response = op.generateResponse( + context.coordinator, + context.lastCommittedOffset + ); + + // The response can be completed immediately. + complete(null); + } catch (Throwable t) { + complete(t); + } + } + + /** + * Completes the future with either the result of the read operation + * or the provided exception. + * + * @param exception The exception to complete the future with. + */ + @Override + public void complete(Throwable exception) { + if (exception == null) { + future.complete(response); + } else { + future.completeExceptionally(exception); + } + } + + @Override + public String toString() { + return "CoordinatorReadEvent(name=" + name + ")"; + } + } + + /** + * A coordinator internal event. + */ + class CoordinatorInternalEvent implements CoordinatorEvent { + /** + * The topic partition that this internal event is applied to. + */ + final TopicPartition tp; + + /** + * The operation name. + */ + final String name; + + /** + * The internal operation to execute. + */ + final Runnable op; + + /** + * Constructor. + * + * @param name The operation name. + * @param tp The topic partition that the operation is applied to. + * @param op The operation. + */ + CoordinatorInternalEvent( + String name, + TopicPartition tp, + Runnable op + ) { + this.tp = tp; + this.name = name; + this.op = op; + } + + /** + * @return The key used by the CoordinatorEventProcessor to ensure + * that events with the same key are not processed concurrently. + */ + @Override + public TopicPartition key() { + return tp; + } + + /** + * Called by the CoordinatorEventProcessor when the event is executed. + */ + @Override + public void run() { + try { + op.run(); + } catch (Throwable t) { + complete(t); + } + } + + /** + * Logs any exceptions thrown while the event is executed. + * + * @param exception The exception. + */ + @Override + public void complete(Throwable exception) { + if (exception != null) { + log.error("Execution of {} failed due to {}.", name, exception); + } + } + + @Override + public String toString() { + return "InternalEvent(name=" + name + ")"; + } + } + + /** + * Partition listener to be notified when the high watermark of the partitions + * backing the coordinator are updated. + */ + class HighWatermarkListener implements PartitionWriter.Listener { + /** + * Updates the high watermark of the corresponding coordinator. + * + * @param tp The topic partition. + * @param offset The new high watermark. + */ + @Override + public void onHighWatermarkUpdated( + TopicPartition tp, + long offset + ) { + log.debug("High watermark of {} incremented to {}.", tp, offset); + scheduleInternalOperation("HighWatermarkUpdated(tp=" + tp + ", offset=" + offset + ")", tp, () -> { + contextOrThrow(tp).updateLastCommittedOffset(offset); + }); + } + } + + /** + * The log context. + */ + private final LogContext logContext; + + /** + * The logger. + */ + private final Logger log; + + /** + * The coordinators keyed by topic partition. + */ + private final ConcurrentHashMap coordinators; + + /** + * The event processor used by the runtime. + */ + private final CoordinatorEventProcessor processor; + + /** + * The partition writer used by the runtime to persist records. + */ + private final PartitionWriter partitionWriter; + + /** + * The high watermark listener registered to all the partitions + * backing the coordinators. + */ + private final HighWatermarkListener highWatermarklistener; + + /** + * The coordinator loaded used by the runtime. + */ + private final CoordinatorLoader loader; + + /** + * The coordinator state machine builder used by the runtime + * to instantiate a coordinator. + */ + private final CoordinatorBuilderSupplier coordinatorBuilderSupplier; + + /** + * Atomic boolean indicating whether the runtime is running. + */ + private final AtomicBoolean isRunning = new AtomicBoolean(true); + + /** + * Constructor. + * + * @param logContext The log context. + * @param processor The event processor. + * @param partitionWriter The partition writer. + * @param loader The coordinator loader. + * @param coordinatorBuilderSupplier The coordinator builder. + */ + private CoordinatorRuntime( + LogContext logContext, + CoordinatorEventProcessor processor, + PartitionWriter partitionWriter, + CoordinatorLoader loader, + CoordinatorBuilderSupplier coordinatorBuilderSupplier + ) { + this.logContext = logContext; + this.log = logContext.logger(CoordinatorRuntime.class); + this.coordinators = new ConcurrentHashMap<>(); + this.processor = processor; + this.partitionWriter = partitionWriter; + this.highWatermarklistener = new HighWatermarkListener(); + this.loader = loader; + this.coordinatorBuilderSupplier = coordinatorBuilderSupplier; + } + + /** + * Throws a NotCoordinatorException exception if the runtime is not + * running. + */ + private void throwIfNotRunning() { + if (!isRunning.get()) { + throw Errors.NOT_COORDINATOR.exception(); + } + } + + /** + * Enqueues a new event. + * + * @param event The event. + * @throws NotCoordinatorException If the event processor is closed. + */ + private void enqueue(CoordinatorEvent event) { + try { + processor.enqueue(event); + } catch (RejectedExecutionException ex) { + throw new NotCoordinatorException("Can't accept an event because the processor is closed", ex); + } + } + + /** + * @return The coordinator context if the coordinator is active or an exception otherwise. + * @throws NotCoordinatorException + * @throws CoordinatorLoadInProgressException + */ + private CoordinatorContext activeContextOrThrow(TopicPartition tp) { + CoordinatorContext context = coordinators.get(tp); + + if (context == null) { + throw Errors.NOT_COORDINATOR.exception(); + } else { + switch (context.state) { + case INITIAL: + case FAILED: + case CLOSED: + throw Errors.NOT_COORDINATOR.exception(); + + case LOADING: + throw Errors.COORDINATOR_LOAD_IN_PROGRESS.exception(); + } + } + + return context; + } + + /** + * @return The coordinator context. It is created if it does not exist. + */ + private CoordinatorContext getOrCreateContext(TopicPartition tp) { + return coordinators.computeIfAbsent(tp, CoordinatorContext::new); + } + + /** + * @return The coordinator context or thrown an exception if it does + * not exist. + * @throws NotCoordinatorException + * Package private for testing. + */ + CoordinatorContext contextOrThrow(TopicPartition tp) { + CoordinatorContext context = coordinators.get(tp); + + if (context == null) { + throw Errors.NOT_COORDINATOR.exception(); + } else { + return context; + } + } + + /** + * Schedules a write operation. + * + * @param name The name of the write operation. + * @param tp The address of the coordinator (aka its topic-partitions). + * @param op The write operation. + * + * @return A future that will be completed with the result of the write operation + * when the operation is completed or an exception if the write operation failed. + * + * @param The type of the result. + */ + public CompletableFuture scheduleWriteOperation( + String name, + TopicPartition tp, + CoordinatorWriteOperation op + ) { + throwIfNotRunning(); + log.debug("Scheduled execution of write operation {}.", name); + CoordinatorWriteEvent event = new CoordinatorWriteEvent<>(name, tp, op); + enqueue(event); + return event.future; + } + + /** + * Schedules a read operation. + * + * @param name The name of the write operation. + * @param tp The address of the coordinator (aka its topic-partitions). + * @param op The read operation. + * + * @return A future that will be completed with the result of the read operation + * when the operation is completed or an exception if the write operation failed. + * + * @param The type of the result. + */ + public CompletableFuture scheduleReadOperation( + String name, + TopicPartition tp, + CoordinatorReadOperation op + ) { + throwIfNotRunning(); + log.debug("Scheduled execution of read operation {}.", name); + CoordinatorReadEvent event = new CoordinatorReadEvent<>(name, tp, op); + enqueue(event); + return event.future; + } + + /** + * Schedules an internal event. + * + * @param name The name of the write operation. + * @param tp The address of the coordinator (aka its topic-partitions). + * @param op The operation. + */ + private void scheduleInternalOperation( + String name, + TopicPartition tp, + Runnable op + ) { + log.debug("Scheduled execution of internal operation {}.", name); + enqueue(new CoordinatorInternalEvent(name, tp, op)); + } + + /** + * @return The topic partitions of the coordinators currently registered in the + * runtime. + */ + public Set partitions() { + throwIfNotRunning(); + return new HashSet<>(coordinators.keySet()); + } + + /** + * Schedules the loading of a coordinator. This is called when the broker is elected as + * the leader for a partition. + * + * @param tp The topic partition of the coordinator. Records from this + * partitions will be read and applied to the coordinator. + * @param partitionEpoch The epoch of the partition. + */ + public void scheduleLoadOperation( + TopicPartition tp, + int partitionEpoch + ) { + throwIfNotRunning(); + log.info("Scheduling loading of metadata from {} with epoch {}", tp, partitionEpoch); + // Touch the state to make the runtime immediately aware of the new coordinator. + getOrCreateContext(tp); + + scheduleInternalOperation("Load(tp=" + tp + ", epoch=" + partitionEpoch + ")", tp, () -> { + CoordinatorContext context = getOrCreateContext(tp); + + if (context.state == CoordinatorState.FAILED) { + // When the coordinator has failed, we create a new context instead of + // recycling the previous one because it is better to start from an + // empty state for timeline data structures. + coordinators.remove(tp); + context = getOrCreateContext(tp); + } + + if (context.epoch < partitionEpoch) { + context.epoch = partitionEpoch; + + switch (context.state) { + case INITIAL: + context.transitionTo(CoordinatorState.LOADING); + loader.load(tp, context.coordinator).whenComplete((state, exception) -> { + scheduleInternalOperation("CompleteLoad(tp=" + tp + ", epoch=" + partitionEpoch + ")", tp, () -> { + CoordinatorContext ctx = contextOrThrow(tp); + if (ctx.state != CoordinatorState.LOADING) { + log.info("Ignoring load completion from {} because context is in {} state.", + ctx.tp, ctx.state); + return; + } + try { + if (exception != null) throw exception; + ctx.transitionTo(CoordinatorState.ACTIVE); + log.info("Finished loading of metadata from {} with epoch {}.", + tp, partitionEpoch); + } catch (Throwable ex) { + log.error("Failed to load metadata from {} with epoch {} due to {}.", + tp, partitionEpoch, ex.toString()); + ctx.transitionTo(CoordinatorState.FAILED); + } + }); + }); + break; + + case LOADING: + log.info("The coordinator {} is already loading metadata.", tp); + break; + + case ACTIVE: + log.info("The coordinator {} is already active.", tp); + break; + + default: + log.error("Cannot load coordinator {} in state {}.", tp, context.state); + } + } else { + log.info("Ignoring loading metadata from {} since current epoch {} is larger than or equals to {}.", + context.tp, context.epoch, partitionEpoch); + } + }); + } + + /** + * Schedules the unloading of a coordinator. This is called when the broker is not the + * leader anymore. + * + * @param tp The topic partition of the coordinator. + * @param partitionEpoch The partition epoch. + */ + public void scheduleUnloadOperation( + TopicPartition tp, + int partitionEpoch + ) { + throwIfNotRunning(); + log.info("Scheduling unloading of metadata for {} with epoch {}", tp, partitionEpoch); + + scheduleInternalOperation("UnloadCoordinator(tp=" + tp + ", epoch=" + partitionEpoch + ")", tp, () -> { + CoordinatorContext context = contextOrThrow(tp); + if (context.epoch < partitionEpoch) { + log.info("Started unloading metadata for {} with epoch {}.", tp, partitionEpoch); + context.transitionTo(CoordinatorState.CLOSED); + coordinators.remove(tp); + log.info("Finished unloading metadata for {} with epoch {}.", tp, partitionEpoch); + } else { + log.info("Ignored unloading metadata for {} in epoch {} since current epoch is {}.", + tp, partitionEpoch, context.epoch); + } + }); + } + + /** + * Closes the runtime. This closes all the coordinators currently registered + * in the runtime. + * + * @throws Exception + */ + public void close() throws Exception { + if (!isRunning.compareAndSet(true, false)) { + log.warn("Coordinator runtime is already shutting down."); + return; + } + + log.info("Closing coordinator runtime."); + // This close the processor, drain all the pending events and + // reject any new events. + processor.close(); + // Unload all the coordinators. + coordinators.forEach((tp, context) -> { + context.transitionTo(CoordinatorState.CLOSED); + }); + coordinators.clear(); + log.info("Coordinator runtime closed."); + } +} diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java index 499ede77fac73..1f35d65c3f443 100644 --- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java +++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.coordinator.group.runtime; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.LogContext; import org.slf4j.Logger; @@ -39,7 +40,7 @@ public class MultiThreadedEventProcessor implements CoordinatorEventProcessor { /** * The accumulator. */ - private final EventAccumulator accumulator; + private final EventAccumulator accumulator; /** * The processing threads. diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java new file mode 100644 index 0000000000000..84e180e32daa1 --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java @@ -0,0 +1,814 @@ +/* + * 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.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.NotCoordinatorException; +import org.apache.kafka.timeline.SnapshotRegistry; +import org.apache.kafka.timeline.TimelineHashSet; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.apache.kafka.common.utils.Utils.mkSet; +import static org.apache.kafka.test.TestUtils.assertFutureThrows; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class CoordinatorRuntimeTest { + private static final TopicPartition TP = new TopicPartition("__consumer_offsets", 0); + + /** + * An CoordinatorEventProcessor that directly executes the operations. This is + * useful in unit tests where execution in threads is not required. + */ + private static class MockEventProcessor implements CoordinatorEventProcessor { + @Override + public void enqueue(CoordinatorEvent event) throws RejectedExecutionException { + try { + event.run(); + } catch (Throwable ex) { + event.complete(ex); + } + } + + @Override + public void close() throws Exception {} + } + + /** + * A CoordinatorLoader that always succeeds. + */ + private static class MockCoordinatorLoader implements CoordinatorLoader { + @Override + public CompletableFuture load(TopicPartition tp, CoordinatorPlayback replayable) { + return CompletableFuture.completedFuture(null); + } + } + + /** + * An in-memory partition writer that accepts a maximum number of writes. + */ + private static class MockPartitionWriter extends InMemoryPartitionWriter { + private final int maxRecordsInBatch; + + public MockPartitionWriter() { + this(Integer.MAX_VALUE); + } + + public MockPartitionWriter(int maxRecordsInBatch) { + super(false); + this.maxRecordsInBatch = maxRecordsInBatch; + } + + @Override + public void registerListener(TopicPartition tp, Listener listener) { + super.registerListener(tp, listener); + } + + @Override + public void deregisterListener(TopicPartition tp, Listener listener) { + super.deregisterListener(tp, listener); + } + + @Override + public long append(TopicPartition tp, List records) throws KafkaException { + if (records.size() <= maxRecordsInBatch) { + return super.append(tp, records); + } else { + throw new KafkaException(String.format("Number of records %d greater than the maximum allowed %d.", + records.size(), maxRecordsInBatch)); + } + } + } + + /** + * A simple Coordinator implementation that stores the records into a set. + */ + private static class MockCoordinator implements Coordinator { + private final TimelineHashSet records; + + MockCoordinator( + SnapshotRegistry snapshotRegistry + ) { + records = new TimelineHashSet<>(snapshotRegistry, 0); + } + + @Override + public void replay(String record) throws RuntimeException { + records.add(record); + } + + Set records() { + return Collections.unmodifiableSet(new HashSet<>(records)); + } + } + + /** + * A CoordinatorBuilder that creates a MockCoordinator. + */ + private static class MockCoordinatorBuilder implements CoordinatorBuilder { + private SnapshotRegistry snapshotRegistry; + + @Override + public CoordinatorBuilder withSnapshotRegistry( + SnapshotRegistry snapshotRegistry + ) { + this.snapshotRegistry = snapshotRegistry; + return this; + } + + @Override + public MockCoordinator build() { + return new MockCoordinator(Objects.requireNonNull(this.snapshotRegistry)); + } + } + + /** + * A CoordinatorBuilderSupplier that returns a MockCoordinatorBuilder. + */ + private static class MockCoordinatorBuilderSupplier implements CoordinatorBuilderSupplier { + @Override + public CoordinatorBuilder get() { + return new MockCoordinatorBuilder(); + } + } + + @Test + public void testScheduleLoading() { + MockCoordinatorLoader loader = mock(MockCoordinatorLoader.class); + MockPartitionWriter writer = mock(MockPartitionWriter.class); + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(loader) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + CompletableFuture future = new CompletableFuture<>(); + when(loader.load(TP, coordinator)).thenReturn(future); + + // Getting the coordinator context fails because the coordinator + // does not exist until scheduleLoadOperation is called. + assertThrows(NotCoordinatorException.class, () -> runtime.contextOrThrow(TP)); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 0); + + // Getting the coordinator context succeeds now. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + + // The coordinator is loading. + assertEquals(CoordinatorRuntime.CoordinatorState.LOADING, ctx.state); + assertEquals(0, ctx.epoch); + assertEquals(coordinator, ctx.coordinator); + + // When the loading completes, the coordinator transitions to active. + future.complete(null); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + + // Verify that onLoaded is called. + verify(coordinator, times(1)).onLoaded(); + + // Verify that the listener is registered. + verify(writer, times(1)).registerListener( + eq(TP), + any(PartitionWriter.Listener.class) + ); + } + + @Test + public void testScheduleLoadingWithFailure() { + MockPartitionWriter writer = mock(MockPartitionWriter.class); + MockCoordinatorLoader loader = mock(MockCoordinatorLoader.class); + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(loader) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + CompletableFuture future = new CompletableFuture<>(); + when(loader.load(TP, coordinator)).thenReturn(future); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 0); + + // Getting the context succeeds and the coordinator should be in loading. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.LOADING, ctx.state); + assertEquals(0, ctx.epoch); + assertEquals(coordinator, ctx.coordinator); + + // When the loading fails, the coordinator transitions to failed. + future.completeExceptionally(new Exception("failure")); + assertEquals(CoordinatorRuntime.CoordinatorState.FAILED, ctx.state); + + // Verify that onUnloaded is called. + verify(coordinator, times(1)).onUnloaded(); + + // Verify that the listener is deregistered. + verify(writer, times(1)).deregisterListener( + eq(TP), + any(PartitionWriter.Listener.class) + ); + } + + @Test + public void testScheduleLoadingWithStalePartitionEpoch() { + MockCoordinatorLoader loader = mock(MockCoordinatorLoader.class); + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(loader) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + CompletableFuture future = new CompletableFuture<>(); + when(loader.load(TP, coordinator)).thenReturn(future); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 10); + + // Getting the context succeeds and the coordinator should be in loading. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.LOADING, ctx.state); + assertEquals(10, ctx.epoch); + assertEquals(coordinator, ctx.coordinator); + + // When the loading completes, the coordinator transitions to active. + future.complete(null); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + assertEquals(10, ctx.epoch); + + // Loading with a previous epoch is a no-op. The coordinator stays + // in active state with the correct epoch. + runtime.scheduleLoadOperation(TP, 0); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + assertEquals(10, ctx.epoch); + } + + @Test + public void testScheduleLoadingAfterLoadingFailure() { + MockCoordinatorLoader loader = mock(MockCoordinatorLoader.class); + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(loader) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + CompletableFuture future = new CompletableFuture<>(); + when(loader.load(TP, coordinator)).thenReturn(future); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 10); + + // Getting the context succeeds and the coordinator should be in loading. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.LOADING, ctx.state); + assertEquals(10, ctx.epoch); + assertEquals(coordinator, ctx.coordinator); + + // When the loading fails, the coordinator transitions to failed. + future.completeExceptionally(new Exception("failure")); + assertEquals(CoordinatorRuntime.CoordinatorState.FAILED, ctx.state); + + // Verify that onUnloaded is called. + verify(coordinator, times(1)).onUnloaded(); + + // Create a new coordinator. + coordinator = mock(MockCoordinator.class); + when(builder.build()).thenReturn(coordinator); + + // Schedule the reloading. + future = new CompletableFuture<>(); + when(loader.load(TP, coordinator)).thenReturn(future); + runtime.scheduleLoadOperation(TP, 11); + + // Getting the context succeeds and the coordinator should be in loading. + ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.LOADING, ctx.state); + assertEquals(11, ctx.epoch); + assertEquals(coordinator, ctx.coordinator); + + // Complete the loading. + future.complete(null); + + // Verify the state. + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + } + + @Test + public void testScheduleUnloading() { + MockPartitionWriter writer = mock(MockPartitionWriter.class); + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + + // Loads the coordinator. It directly transitions to active. + runtime.scheduleLoadOperation(TP, 10); + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + assertEquals(10, ctx.epoch); + + // Schedule the unloading. + runtime.scheduleUnloadOperation(TP, ctx.epoch + 1); + assertEquals(CoordinatorRuntime.CoordinatorState.CLOSED, ctx.state); + + // Verify that onUnloaded is called. + verify(coordinator, times(1)).onUnloaded(); + + // Verify that the listener is deregistered. + verify(writer, times(1)).deregisterListener( + eq(TP), + any(PartitionWriter.Listener.class) + ); + + // Getting the coordinator context fails because it no longer exists. + assertThrows(NotCoordinatorException.class, () -> runtime.contextOrThrow(TP)); + } + + @Test + public void testScheduleUnloadingWithStalePartitionEpoch() { + MockCoordinatorBuilderSupplier supplier = mock(MockCoordinatorBuilderSupplier.class); + MockCoordinatorBuilder builder = mock(MockCoordinatorBuilder.class); + MockCoordinator coordinator = mock(MockCoordinator.class); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(supplier) + .build(); + + when(builder.withSnapshotRegistry(any())).thenReturn(builder); + when(builder.build()).thenReturn(coordinator); + when(supplier.get()).thenReturn(builder); + + // Loads the coordinator. It directly transitions to active. + runtime.scheduleLoadOperation(TP, 10); + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + assertEquals(10, ctx.epoch); + + // Unloading with a previous epoch is a no-op. The coordinator stays + // in active with the correct epoch. + runtime.scheduleUnloadOperation(TP, 0); + assertEquals(CoordinatorRuntime.CoordinatorState.ACTIVE, ctx.state); + assertEquals(10, ctx.epoch); + } + + @Test + public void testScheduleWriteOp() throws ExecutionException, InterruptedException, TimeoutException { + MockPartitionWriter writer = new MockPartitionWriter(); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Schedule the loading. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Collections.singletonList(0L), ctx.snapshotRegistry.epochsList()); + + // Write #1. + CompletableFuture write1 = runtime.scheduleWriteOperation("write#1", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + + // Verify that the write is not committed yet. + assertFalse(write1.isDone()); + + // The last written offset is updated. + assertEquals(2L, ctx.lastWrittenOffset); + // The last committed offset does not change. + assertEquals(0L, ctx.lastCommittedOffset); + // A new snapshot is created. + assertEquals(Arrays.asList(0L, 2L), ctx.snapshotRegistry.epochsList()); + // Records have been replayed to the coordinator. + assertEquals(mkSet("record1", "record2"), ctx.coordinator.records()); + // Records have been written to the log. + assertEquals(Arrays.asList("record1", "record2"), writer.records(TP)); + + // Write #2. + CompletableFuture write2 = runtime.scheduleWriteOperation("write#2", TP, + state -> new CoordinatorResult<>(Arrays.asList("record3"), "response2")); + + // Verify that the write is not committed yet. + assertFalse(write2.isDone()); + + // The last written offset is updated. + assertEquals(3L, ctx.lastWrittenOffset); + // The last committed offset does not change. + assertEquals(0L, ctx.lastCommittedOffset); + // A new snapshot is created. + assertEquals(Arrays.asList(0L, 2L, 3L), ctx.snapshotRegistry.epochsList()); + // Records have been replayed to the coordinator. + assertEquals(mkSet("record1", "record2", "record3"), ctx.coordinator.records()); + // Records have been written to the log. + assertEquals(Arrays.asList("record1", "record2", "record3"), writer.records(TP)); + + // Write #3 but without any records. + CompletableFuture write3 = runtime.scheduleWriteOperation("write#3", TP, + state -> new CoordinatorResult<>(Collections.emptyList(), "response3")); + + // Verify that the write is not committed yet. + assertFalse(write3.isDone()); + + // The state does not change. + assertEquals(3L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Arrays.asList(0L, 2L, 3L), ctx.snapshotRegistry.epochsList()); + assertEquals(mkSet("record1", "record2", "record3"), ctx.coordinator.records()); + assertEquals(Arrays.asList("record1", "record2", "record3"), writer.records(TP)); + + // Commit write #1. + writer.commit(TP, 2); + + // The write is completed. + assertTrue(write1.isDone()); + assertEquals("response1", write1.get(5, TimeUnit.SECONDS)); + + // The last committed offset is updated. + assertEquals(2L, ctx.lastCommittedOffset); + // The snapshot is cleaned up. + assertEquals(Arrays.asList(2L, 3L), ctx.snapshotRegistry.epochsList()); + + // Commit write #2. + writer.commit(TP, 3); + + // The writes are completed. + assertTrue(write2.isDone()); + assertTrue(write3.isDone()); + assertEquals("response2", write2.get(5, TimeUnit.SECONDS)); + assertEquals("response3", write3.get(5, TimeUnit.SECONDS)); + + // The last committed offset is updated. + assertEquals(3L, ctx.lastCommittedOffset); + // The snapshot is cleaned up. + assertEquals(Collections.singletonList(3L), ctx.snapshotRegistry.epochsList()); + + // Write #4 but without records. + CompletableFuture write4 = runtime.scheduleWriteOperation("write#4", TP, + state -> new CoordinatorResult<>(Collections.emptyList(), "response4")); + + // It is completed immediately because the state is fully commited. + assertTrue(write4.isDone()); + assertEquals("response4", write4.get(5, TimeUnit.SECONDS)); + assertEquals(Collections.singletonList(3L), ctx.snapshotRegistry.epochsList()); + } + + @Test + public void testScheduleWriteOpWhenInactive() { + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Scheduling a write fails with a NotCoordinatorException because the coordinator + // does not exist. + CompletableFuture write = runtime.scheduleWriteOperation("write", TP, + state -> new CoordinatorResult<>(Collections.emptyList(), "response1")); + assertFutureThrows(write, NotCoordinatorException.class); + } + + @Test + public void testScheduleWriteOpWhenOpFails() { + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Scheduling a write that fails when the operation is called. The exception + // is used to complete the future. + CompletableFuture write = runtime.scheduleWriteOperation("write", TP, state -> { + throw new KafkaException("error"); + }); + assertFutureThrows(write, KafkaException.class); + } + + @Test + public void testScheduleWriteOpWhenReplayFails() { + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Collections.singletonList(0L), ctx.snapshotRegistry.epochsList()); + + // Override the coordinator with a coordinator that throws + // an exception when replay is called. + ctx.coordinator = new MockCoordinator(ctx.snapshotRegistry) { + @Override + public void replay(String record) throws RuntimeException { + throw new IllegalArgumentException("error"); + } + }; + + // Write. It should fail. + CompletableFuture write = runtime.scheduleWriteOperation("write", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + assertFutureThrows(write, IllegalArgumentException.class); + + // Verify that the state has not changed. + assertEquals(0L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Collections.singletonList(0L), ctx.snapshotRegistry.epochsList()); + } + + @Test + public void testScheduleWriteOpWhenWriteFails() { + // The partition writer only accept on write. + MockPartitionWriter writer = new MockPartitionWriter(2); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0, ctx.lastWrittenOffset); + assertEquals(0, ctx.lastCommittedOffset); + assertEquals(Collections.singletonList(0L), ctx.snapshotRegistry.epochsList()); + + // Write #1. It should succeed and be applied to the coordinator. + runtime.scheduleWriteOperation("write#1", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + + // Verify that the state has been updated. + assertEquals(2L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Arrays.asList(0L, 2L), ctx.snapshotRegistry.epochsList()); + assertEquals(mkSet("record1", "record2"), ctx.coordinator.records()); + + // Write #2. It should fail because the writer is configured to only + // accept 2 records per batch. + CompletableFuture write2 = runtime.scheduleWriteOperation("write#2", TP, + state -> new CoordinatorResult<>(Arrays.asList("record3", "record4", "record5"), "response2")); + assertFutureThrows(write2, KafkaException.class); + + // Verify that the state has not changed. + assertEquals(2L, ctx.lastWrittenOffset); + assertEquals(0L, ctx.lastCommittedOffset); + assertEquals(Arrays.asList(0L, 2L), ctx.snapshotRegistry.epochsList()); + assertEquals(mkSet("record1", "record2"), ctx.coordinator.records()); + } + + @Test + public void testScheduleReadOp() throws ExecutionException, InterruptedException, TimeoutException { + MockPartitionWriter writer = new MockPartitionWriter(); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0, ctx.lastWrittenOffset); + assertEquals(0, ctx.lastCommittedOffset); + + // Write #1. + CompletableFuture write1 = runtime.scheduleWriteOperation("write#1", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + + // Write #2. + CompletableFuture write2 = runtime.scheduleWriteOperation("write#2", TP, + state -> new CoordinatorResult<>(Arrays.asList("record3", "record4"), "response2")); + + // Commit write #1. + writer.commit(TP, 2); + + // Write #1 is completed. + assertTrue(write1.isDone()); + + // Write #2 is not. + assertFalse(write2.isDone()); + + // The last written and committed offsets are updated. + assertEquals(4, ctx.lastWrittenOffset); + assertEquals(2, ctx.lastCommittedOffset); + + // Read. + CompletableFuture read = runtime.scheduleReadOperation("read", TP, (state, offset) -> { + // The read operation should be given the last committed offset. + assertEquals(ctx.lastCommittedOffset, offset); + return "read-response"; + }); + + // The read is completed immediately. + assertTrue(read.isDone()); + assertEquals("read-response", read.get(5, TimeUnit.SECONDS)); + } + + @Test + public void testScheduleReadOpWhenPartitionInactive() { + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Schedule a read. It fails because the coordinator does not exist. + CompletableFuture read = runtime.scheduleReadOperation("read", TP, + (state, offset) -> "read-response"); + assertFutureThrows(read, NotCoordinatorException.class); + } + + @Test + public void testScheduleReadOpWhenOpsFails() { + MockPartitionWriter writer = new MockPartitionWriter(); + + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(writer) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Verify the initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0, ctx.lastWrittenOffset); + assertEquals(0, ctx.lastCommittedOffset); + + // Write #1. + runtime.scheduleWriteOperation("write#1", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + + // Write #2. + runtime.scheduleWriteOperation("write#2", TP, + state -> new CoordinatorResult<>(Arrays.asList("record3", "record4"), "response2")); + + // Commit write #1. + writer.commit(TP, 2); + + // Read. It fails with an exception that is used to complete the future. + CompletableFuture read = runtime.scheduleReadOperation("read", TP, (state, offset) -> { + assertEquals(ctx.lastCommittedOffset, offset); + throw new IllegalArgumentException("error"); + }); + assertFutureThrows(read, IllegalArgumentException.class); + } + + @Test + public void testClose() throws Exception { + CoordinatorRuntime runtime = + new CoordinatorRuntime.Builder() + .withLoader(new MockCoordinatorLoader()) + .withEventProcessor(new MockEventProcessor()) + .withPartitionWriter(new MockPartitionWriter()) + .withCoordinatorBuilderSupplier(new MockCoordinatorBuilderSupplier()) + .build(); + + // Loads the coordinator. + runtime.scheduleLoadOperation(TP, 10); + + // Check initial state. + CoordinatorRuntime.CoordinatorContext ctx = runtime.contextOrThrow(TP); + assertEquals(0, ctx.lastWrittenOffset); + assertEquals(0, ctx.lastCommittedOffset); + + // Write #1. + CompletableFuture write1 = runtime.scheduleWriteOperation("write#1", TP, + state -> new CoordinatorResult<>(Arrays.asList("record1", "record2"), "response1")); + + // Write #2. + CompletableFuture write2 = runtime.scheduleWriteOperation("write#2", TP, + state -> new CoordinatorResult<>(Arrays.asList("record3", "record4"), "response2")); + + // Writes are inflight. + assertFalse(write1.isDone()); + assertFalse(write2.isDone()); + + // Close the runtime. + runtime.close(); + + // All the pending operations are completed with NotCoordinatorException. + assertFutureThrows(write1, NotCoordinatorException.class); + assertFutureThrows(write2, NotCoordinatorException.class); + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java new file mode 100644 index 0000000000000..f31b4b27a4dcc --- /dev/null +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/InMemoryPartitionWriter.java @@ -0,0 +1,145 @@ +/* + * 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.common.KafkaException; +import org.apache.kafka.common.TopicPartition; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; + +/** + * An in-memory partition writer. + * + * @param The record type. + */ +public class InMemoryPartitionWriter implements PartitionWriter { + + private class PartitionState { + private ReentrantLock lock = new ReentrantLock(); + private List listeners = new ArrayList<>(); + private List records = new ArrayList<>(); + private long endOffset = 0L; + private long committedOffset = 0L; + } + + private final boolean autoCommit; + private final Map partitions; + + public InMemoryPartitionWriter(boolean autoCommit) { + this.autoCommit = autoCommit; + this.partitions = new ConcurrentHashMap<>(); + } + + private PartitionState partitionState( + TopicPartition tp + ) { + return partitions.computeIfAbsent(tp, __ -> new PartitionState()); + } + + @Override + public void registerListener( + TopicPartition tp, + Listener listener + ) { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + state.listeners.add(listener); + } finally { + state.lock.unlock(); + } + } + + @Override + public void deregisterListener( + TopicPartition tp, + Listener listener + ) { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + state.listeners.remove(listener); + } finally { + state.lock.unlock(); + } + } + + @Override + public long append( + TopicPartition tp, + List records + ) throws KafkaException { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + state.records.addAll(records); + state.endOffset += records.size(); + if (autoCommit) commit(tp, state.endOffset); + return state.endOffset; + } finally { + state.lock.unlock(); + } + } + + public void commit( + TopicPartition tp, + long offset + ) { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + state.committedOffset = offset; + state.listeners.forEach(listener -> { + listener.onHighWatermarkUpdated(tp, state.committedOffset); + }); + } finally { + state.lock.unlock(); + } + } + + public void commit( + TopicPartition tp + ) { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + state.committedOffset = state.endOffset; + state.listeners.forEach(listener -> { + listener.onHighWatermarkUpdated(tp, state.committedOffset); + }); + } finally { + state.lock.unlock(); + } + } + + public List records( + TopicPartition tp + ) { + PartitionState state = partitionState(tp); + state.lock.lock(); + try { + return Collections.unmodifiableList(state.records); + } finally { + state.lock.unlock(); + } + } +} diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java index c11248b187534..0630c88f38995 100644 --- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java +++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.coordinator.group.runtime; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.LogContext; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; @@ -41,7 +42,7 @@ public class MultiThreadedEventProcessorTest { private static class FutureEvent implements CoordinatorEvent { - private final int key; + private final TopicPartition key; private final CompletableFuture future; private final Supplier supplier; private final boolean block; @@ -49,14 +50,14 @@ private static class FutureEvent implements CoordinatorEvent { private final CountDownLatch executed; FutureEvent( - int key, + TopicPartition key, Supplier supplier ) { this(key, supplier, false); } FutureEvent( - int key, + TopicPartition key, Supplier supplier, boolean block ) { @@ -90,7 +91,7 @@ public void complete(Throwable ex) { } @Override - public Integer key() { + public TopicPartition key() { return key; } @@ -132,12 +133,12 @@ public void testEventsAreProcessed() throws Exception { AtomicInteger numEventsExecuted = new AtomicInteger(0); List> events = Arrays.asList( - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(1, numEventsExecuted::incrementAndGet), - new FutureEvent<>(2, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(1, numEventsExecuted::incrementAndGet), - new FutureEvent<>(2, numEventsExecuted::incrementAndGet) + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 2), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 2), numEventsExecuted::incrementAndGet) ); events.forEach(eventProcessor::enqueue); @@ -167,12 +168,12 @@ public void testProcessingGuarantees() throws Exception { AtomicInteger numEventsExecuted = new AtomicInteger(0); List> events = Arrays.asList( - new FutureEvent<>(0, numEventsExecuted::incrementAndGet, true), // Event 0 - new FutureEvent<>(1, numEventsExecuted::incrementAndGet, true), // Event 1 - new FutureEvent<>(0, numEventsExecuted::incrementAndGet, true), // Event 2 - new FutureEvent<>(1, numEventsExecuted::incrementAndGet, true), // Event 3 - new FutureEvent<>(0, numEventsExecuted::incrementAndGet, true), // Event 4 - new FutureEvent<>(1, numEventsExecuted::incrementAndGet, true) // Event 5 + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet, true), // Event 0 + new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet, true), // Event 1 + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet, true), // Event 2 + new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet, true), // Event 3 + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet, true), // Event 4 + new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet, true) // Event 5 ); events.forEach(eventProcessor::enqueue); @@ -251,7 +252,7 @@ public void testEventsAreRejectedWhenClosed() throws Exception { eventProcessor.close(); assertThrows(RejectedExecutionException.class, - () -> eventProcessor.enqueue(new FutureEvent<>(0, () -> 0))); + () -> eventProcessor.enqueue(new FutureEvent<>(new TopicPartition("foo", 0), () -> 0))); } @Test @@ -264,15 +265,15 @@ public void testEventsAreDrainedWhenClosed() throws Exception { AtomicInteger numEventsExecuted = new AtomicInteger(0); // Special event which blocks until the latch is released. - FutureEvent blockingEvent = new FutureEvent<>(0, numEventsExecuted::incrementAndGet, true); + FutureEvent blockingEvent = new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet, true); List> events = Arrays.asList( - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet), - new FutureEvent<>(0, numEventsExecuted::incrementAndGet) + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet), + new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet) ); // Enqueue the blocking event.