diff --git a/checkstyle/import-control-metadata.xml b/checkstyle/import-control-metadata.xml index c2660674e3259..293801bd75f2b 100644 --- a/checkstyle/import-control-metadata.xml +++ b/checkstyle/import-control-metadata.xml @@ -83,7 +83,6 @@ - @@ -160,7 +159,6 @@ - @@ -198,18 +196,4 @@ - - - - - - - - - - - - - - diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java index 5f4d6142434ae..3e1dd69723b8d 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java +++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java @@ -260,8 +260,8 @@ public Builder setConfigSchema(KafkaConfigSchema configSchema) { return this; } - public Builder setRaftClient(RaftClient logManager) { - this.raftClient = logManager; + public Builder setRaftClient(RaftClient raftClient) { + this.raftClient = raftClient; return this; } @@ -1082,7 +1082,7 @@ public void handleLeaderChange(LeaderAndEpoch newLeader) { @Override public void beginShutdown() { - queue.beginShutdown("MetaLogManager.Listener"); + queue.beginShutdown("QuorumMetaLogListener"); } private void appendRaftEvent(String name, Runnable runnable) { diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java similarity index 91% rename from metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java rename to metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java index 013d5945d2ea2..21e3a64553042 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.metalog; +package org.apache.kafka.controller; import org.apache.kafka.common.protocol.ObjectSerializationCache; import org.apache.kafka.common.utils.BufferSupplier; @@ -66,9 +66,9 @@ import java.util.concurrent.atomic.AtomicBoolean; /** - * The LocalLogManager is a test implementation that relies on the contents of memory. + * The MockRaftClient is a test implementation that relies on the contents of memory. */ -public final class LocalLogManager implements RaftClient, AutoCloseable { +public final class MockRaftClient implements RaftClient, AutoCloseable { interface LocalBatch { int epoch(); int size(); @@ -158,9 +158,9 @@ public static class SharedLogData { private static final Logger log = LoggerFactory.getLogger(SharedLogData.class); /** - * Maps node IDs to the matching log managers. + * Maps node IDs to the matching raft clients. */ - private final HashMap logManagers = new HashMap<>(); + private final HashMap raftClients = new HashMap<>(); /** * Maps offsets to record batches. @@ -198,17 +198,17 @@ public SharedLogData(Optional snapshot) { } } - synchronized void registerLogManager(LocalLogManager logManager) { - if (logManagers.put(logManager.nodeId, logManager) != null) { - throw new RuntimeException("Can't have multiple LocalLogManagers " + - "with id " + logManager.nodeId()); + synchronized void registerRaftClient(MockRaftClient raftClient) { + if (raftClients.put(raftClient.nodeId, raftClient) != null) { + throw new RuntimeException("Can't have multiple MockRaftClients " + + "with id " + raftClient.nodeId()); } electLeaderIfNeeded(); } - synchronized void unregisterLogManager(LocalLogManager logManager) { - if (!logManagers.remove(logManager.nodeId, logManager)) { - throw new RuntimeException("Log manager " + logManager.nodeId() + + synchronized void unregisterRaftClient(MockRaftClient raftClient) { + if (!raftClients.remove(raftClient.nodeId, raftClient)) { + throw new RuntimeException("MockRaftClient " + raftClient.nodeId() + " was not found."); } } @@ -259,19 +259,19 @@ public synchronized long append( if (batch instanceof LeaderChangeBatch leaderChangeBatch) { leader = leaderChangeBatch.newLeader; } - for (LocalLogManager logManager : logManagers.values()) { - logManager.scheduleLogCheck(); + for (MockRaftClient raftClient : raftClients.values()) { + raftClient.scheduleLogCheck(); } prevOffset = nextEndOffset; return nextEndOffset; } synchronized void electLeaderIfNeeded() { - if (leader.leaderId().isPresent() || logManagers.isEmpty()) { + if (leader.leaderId().isPresent() || raftClients.isEmpty()) { return; } - int nextLeaderIndex = ThreadLocalRandom.current().nextInt(logManagers.size()); - Iterator iter = logManagers.keySet().iterator(); + int nextLeaderIndex = ThreadLocalRandom.current().nextInt(raftClients.size()); + Iterator iter = raftClients.keySet().iterator(); Integer nextLeaderNode = null; for (int i = 0; i <= nextLeaderIndex; i++) { nextLeaderNode = iter.next(); @@ -294,7 +294,7 @@ synchronized Entry nextBatch(long offset) { } /** - * Optionally return a snapshot reader if the offset if less than the first batch. + * Optionally return a snapshot reader if the offset is less than the first batch. */ synchronized Optional nextSnapshot(long offset) { return Optional.ofNullable(snapshots.lastEntry()).flatMap(entry -> { @@ -325,7 +325,7 @@ synchronized void addSnapshot(RawSnapshotReader newSnapshot) { /** * Returns the snapshot id of the latest snapshot if there is one. * - * If a snapshot doesn't exists, it return an empty Optional. + * If a snapshot doesn't exist, it returns an empty Optional. */ synchronized Optional latestSnapshotId() { return Optional.ofNullable(snapshots.lastEntry()).map(entry -> entry.getValue().snapshotId()); @@ -400,65 +400,65 @@ void beginShutdown() { private final Logger log; /** - * The node ID of this local log manager. Each log manager must have a unique ID. + * The node ID of this raft client. Each raft client must have a unique ID. */ private final int nodeId; /** - * A reference to the in-memory state that unites all the log managers in use. + * A reference to the in-memory state that unites all the raft clients in use. */ private final SharedLogData shared; /** - * The event queue used by this local log manager. + * The event queue used by this raft client. */ private final EventQueue eventQueue; /** - * The latest kraft version used by this local log manager. + * The latest kraft version used by this raft client. */ private KRaftVersion lastKRaftVersion; /** - * Whether this LocalLogManager has been shut down. + * Whether this raft client has been shut down. */ private boolean shutdown = false; /** - * An offset that the log manager will not read beyond. This exists only for testing + * An offset that the raft client will not read beyond. This exists only for testing * purposes. */ private long maxReadOffset; /** - * The listener objects attached to this local log manager. + * The listener objects attached to this raft client. */ private final Map, MetaLogListenerData> listeners = new IdentityHashMap<>(); /** - * The current leader, as seen by this log manager. + * The current leader, as seen by this raft client. */ private volatile LeaderAndEpoch leader = new LeaderAndEpoch(OptionalInt.empty(), 0); - /* + /** * If this variable is true the next scheduleAppend will fail */ private final AtomicBoolean throwOnNextAppend = new AtomicBoolean(false); - public LocalLogManager(LogContext logContext, - int nodeId, - SharedLogData shared, - String threadNamePrefix, - KRaftVersion lastKRaftVersion) { + public MockRaftClient(LogContext logContext, + int nodeId, + SharedLogData shared, + String threadNamePrefix, + KRaftVersion lastKRaftVersion) { this.logContext = logContext; - this.log = logContext.logger(LocalLogManager.class); + this.log = logContext.logger(MockRaftClient.class); this.nodeId = nodeId; this.shared = shared; this.maxReadOffset = shared.initialMaxReadOffset(); this.eventQueue = new KafkaEventQueue(Time.SYSTEM, logContext, threadNamePrefix, new ShutdownEvent()); this.lastKRaftVersion = lastKRaftVersion; - shared.registerLogManager(this); + this.shared.registerRaftClient(this); } private void scheduleLogCheck() { @@ -477,7 +477,7 @@ private void scheduleLogCheck() { listenerData.handleLoadSnapshot( RecordsSnapshotReader.of( snapshot.get(), - new MetadataRecordSerde(), + new MetadataRecordSerde(), BufferSupplier.create(), Integer.MAX_VALUE, true, @@ -556,7 +556,7 @@ private static int messageSize(ApiMessageAndVersion messageAndVersion, ObjectSer } public void beginShutdown() { - eventQueue.beginShutdown("beginShutdown"); + eventQueue.beginShutdown("MockKafkaRaftClient"); } class ShutdownEvent implements EventQueue.Event { @@ -569,7 +569,7 @@ public void run() throws Exception { for (MetaLogListenerData listenerData : listeners.values()) { listenerData.beginShutdown(); } - shared.unregisterLogManager(LocalLogManager.this); + shared.unregisterRaftClient(MockRaftClient.this); } } catch (Exception e) { log.error("Unexpected exception while sending beginShutdown callbacks", e); @@ -592,7 +592,7 @@ public void close() { } /** - * Shutdown the log manager. + * Shutdown the raft client. * * Even though the API suggests a non-blocking shutdown, this method always returns a completed * future. This means that shutdown is a blocking operation. @@ -614,7 +614,7 @@ public void register(RaftClient.Listener listener) { CompletableFuture future = new CompletableFuture<>(); eventQueue.append(() -> { if (shutdown) { - log.info("Node {}: can't register because local log manager has " + + log.info("Node {}: can't register because raft client has " + "already been shut down.", nodeId); future.complete(null); } else { @@ -643,7 +643,7 @@ public void register(RaftClient.Listener listener) { public void unregister(RaftClient.Listener listener) { eventQueue.append(() -> { if (shutdown) { - log.info("Node {}: can't unregister because local log manager is shutdown", nodeId); + log.info("Node {}: can't unregister because raft client is shutdown", nodeId); } else { int id = System.identityHashCode(listener); if (listeners.remove(listener) == null) { diff --git a/metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientListener.java similarity index 95% rename from metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java rename to metadata/src/test/java/org/apache/kafka/controller/MockRaftClientListener.java index 53bc49f807531..b4ee74346ee98 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientListener.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.metalog; +package org.apache.kafka.controller; import org.apache.kafka.common.protocol.ApiMessage; import org.apache.kafka.raft.Batch; @@ -29,7 +29,7 @@ import java.util.List; import java.util.OptionalInt; -public class MockMetaLogManagerListener implements RaftClient.Listener { +public class MockRaftClientListener implements RaftClient.Listener { public static final String COMMIT = "COMMIT"; public static final String LAST_COMMITTED_OFFSET = "LAST_COMMITTED_OFFSET"; public static final String NEW_LEADER = "NEW_LEADER"; @@ -41,7 +41,7 @@ public class MockMetaLogManagerListener implements RaftClient.Listener serializedEvents = new ArrayList<>(); private LeaderAndEpoch leaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty(), 0); - public MockMetaLogManagerListener(int nodeId) { + public MockRaftClientListener(int nodeId) { this.nodeId = nodeId; } diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTest.java similarity index 72% rename from metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java rename to metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTest.java index b42f781e98551..4217bd4ed2602 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTest.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.kafka.metalog; +package org.apache.kafka.controller; import org.apache.kafka.common.metadata.RegisterBrokerRecord; import org.apache.kafka.raft.LeaderAndEpoch; @@ -28,23 +28,23 @@ import java.util.List; import java.util.OptionalInt; -import static org.apache.kafka.metalog.MockMetaLogManagerListener.COMMIT; -import static org.apache.kafka.metalog.MockMetaLogManagerListener.LAST_COMMITTED_OFFSET; -import static org.apache.kafka.metalog.MockMetaLogManagerListener.SHUTDOWN; +import static org.apache.kafka.controller.MockRaftClientListener.COMMIT; +import static org.apache.kafka.controller.MockRaftClientListener.LAST_COMMITTED_OFFSET; +import static org.apache.kafka.controller.MockRaftClientListener.SHUTDOWN; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; @Timeout(value = 40) -public class LocalLogManagerTest { +public class MockRaftClientTest { /** - * Test creating a LocalLogManager and closing it. + * Test creating a MockRaftClient and closing it. */ @Test public void testCreateAndClose() throws Exception { try ( - LocalLogManagerTestEnv env = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv env = new MockRaftClientTestEnv.Builder(1). buildWithMockListeners() ) { env.close(); @@ -53,12 +53,12 @@ public void testCreateAndClose() throws Exception { } /** - * Test that the local log manager will claim leadership. + * Test that the raft client will claim leadership. */ @Test public void testClaimsLeadership() throws Exception { try ( - LocalLogManagerTestEnv env = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv env = new MockRaftClientTestEnv.Builder(1). buildWithMockListeners() ) { assertEquals(new LeaderAndEpoch(OptionalInt.of(0), 1), env.waitForLeader()); @@ -68,12 +68,12 @@ public void testClaimsLeadership() throws Exception { } /** - * Test that we can pass leadership back and forth between log managers. + * Test that we can pass leadership back and forth between raft clients. */ @Test public void testPassLeadership() throws Exception { try ( - LocalLogManagerTestEnv env = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv env = new MockRaftClientTestEnv.Builder(3). buildWithMockListeners() ) { LeaderAndEpoch first = env.waitForLeader(); @@ -82,7 +82,7 @@ public void testPassLeadership() throws Exception { int currentLeaderId = cur.leaderId().orElseThrow(() -> new AssertionError("Current leader is undefined") ); - env.logManagers().get(currentLeaderId).resign(cur.epoch()); + env.raftClients().get(currentLeaderId).resign(cur.epoch()); LeaderAndEpoch next = env.waitForLeader(); while (next.epoch() == cur.epoch()) { @@ -100,10 +100,9 @@ public void testPassLeadership() throws Exception { } private static void waitForLastCommittedOffset(long targetOffset, - LocalLogManager logManager) throws InterruptedException { + MockRaftClient raftClient) throws InterruptedException { TestUtils.retryOnExceptionWithTimeout(20000, 3, () -> { - MockMetaLogManagerListener listener = - (MockMetaLogManagerListener) logManager.listeners().get(0); + MockRaftClientListener listener = (MockRaftClientListener) raftClient.listeners().get(0); long highestOffset = -1; for (String event : listener.serializedEvents()) { if (event.startsWith(LAST_COMMITTED_OFFSET)) { @@ -117,19 +116,19 @@ private static void waitForLastCommittedOffset(long targetOffset, } } if (highestOffset < targetOffset) { - throw new RuntimeException("Offset for log manager " + - logManager.nodeId() + " only reached " + highestOffset); + throw new RuntimeException("Offset for raft client " + + raftClient.nodeId() + " only reached " + highestOffset); } }); } /** - * Test that all the log managers see all the commits. + * Test that all the raft clients see all the commits. */ @Test public void testCommits() throws Exception { try ( - LocalLogManagerTestEnv env = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv env = new MockRaftClientTestEnv.Builder(3). buildWithMockListeners() ) { LeaderAndEpoch leaderInfo = env.waitForLeader(); @@ -137,22 +136,24 @@ public void testCommits() throws Exception { new AssertionError("Current leader is undefined") ); - LocalLogManager activeLogManager = env.logManagers().get(leaderId); - int epoch = activeLogManager.leaderAndEpoch().epoch(); + MockRaftClient activeRaftClient = env.raftClients().get(leaderId); + int epoch = activeRaftClient.leaderAndEpoch().epoch(); List messages = List.of( new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(0), (short) 0), new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(1), (short) 0), new ApiMessageAndVersion(new RegisterBrokerRecord().setBrokerId(2), (short) 0)); - assertEquals(3, activeLogManager.prepareAppend(epoch, messages)); - activeLogManager.schedulePreparedAppend(); - for (LocalLogManager logManager : env.logManagers()) { - waitForLastCommittedOffset(3, logManager); + assertEquals(3, activeRaftClient.prepareAppend(epoch, messages)); + + activeRaftClient.schedulePreparedAppend(); + for (MockRaftClient raftClient : env.raftClients()) { + waitForLastCommittedOffset(3, raftClient); } - List listeners = env.logManagers().stream(). - map(m -> (MockMetaLogManagerListener) m.listeners().get(0)). + + List listeners = env.raftClients().stream(). + map(m -> (MockRaftClientListener) m.listeners().get(0)). toList(); env.close(); - for (MockMetaLogManagerListener listener : listeners) { + for (MockRaftClientListener listener : listeners) { List events = listener.serializedEvents(); assertEquals(SHUTDOWN, events.get(events.size() - 1)); int foundIndex = 0; diff --git a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTestEnv.java similarity index 73% rename from metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java rename to metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTestEnv.java index a1e6742f8efe7..6d1e741d1d82d 100644 --- a/metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java +++ b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClientTestEnv.java @@ -15,14 +15,14 @@ * limitations under the License. */ -package org.apache.kafka.metalog; +package org.apache.kafka.controller; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Utils; -import org.apache.kafka.metalog.LocalLogManager.LeaderChangeBatch; -import org.apache.kafka.metalog.LocalLogManager.LocalRecordBatch; -import org.apache.kafka.metalog.LocalLogManager.SharedLogData; +import org.apache.kafka.controller.MockRaftClient.LeaderChangeBatch; +import org.apache.kafka.controller.MockRaftClient.LocalRecordBatch; +import org.apache.kafka.controller.MockRaftClient.SharedLogData; import org.apache.kafka.raft.LeaderAndEpoch; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.server.common.KRaftVersion; @@ -41,9 +41,9 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; -public class LocalLogManagerTestEnv implements AutoCloseable { +public class MockRaftClientTestEnv implements AutoCloseable { private static final Logger log = - LoggerFactory.getLogger(LocalLogManagerTestEnv.class); + LoggerFactory.getLogger(MockRaftClientTestEnv.class); private final String clusterId; @@ -59,23 +59,23 @@ public class LocalLogManagerTestEnv implements AutoCloseable { private final File dir; /** - * The shared data for our LocalLogManager instances. + * The shared data for our MockRaftClient instances. */ private final SharedLogData shared; /** - * A list of log managers. + * A list of raft clients. */ - private final List logManagers; + private final List raftClients; public static class Builder { - private final int numManagers; + private final int numNodes; private Optional snapshotReader = Optional.empty(); private Consumer sharedLogDataInitializer = __ -> { }; private KRaftVersion lastKRaftVersion = KRaftVersion.KRAFT_VERSION_0; - public Builder(int numManagers) { - this.numManagers = numManagers; + public Builder(int numNodes) { + this.numNodes = numNodes; } public Builder setSnapshotReader(RawSnapshotReader snapshotReader) { @@ -96,19 +96,19 @@ public Builder setLastKRaftVersion(KRaftVersion kraftVersion) { return this; } - public LocalLogManagerTestEnv build() { - return new LocalLogManagerTestEnv( - numManagers, + public MockRaftClientTestEnv build() { + return new MockRaftClientTestEnv( + numNodes, snapshotReader, sharedLogDataInitializer, lastKRaftVersion); } - public LocalLogManagerTestEnv buildWithMockListeners() { - LocalLogManagerTestEnv env = build(); + public MockRaftClientTestEnv buildWithMockListeners() { + MockRaftClientTestEnv env = build(); try { - for (LocalLogManager logManager : env.logManagers) { - logManager.register(new MockMetaLogManagerListener(logManager.nodeId().getAsInt())); + for (MockRaftClient raftClient : env.raftClients) { + raftClient.register(new MockRaftClientListener(raftClient.nodeId().getAsInt())); } } catch (Exception e) { try { @@ -122,8 +122,8 @@ public LocalLogManagerTestEnv buildWithMockListeners() { } } - private LocalLogManagerTestEnv( - int numManagers, + private MockRaftClientTestEnv( + int numNodes, Optional snapshotReader, Consumer sharedLogDataInitializer, KRaftVersion lastKRaftVersion @@ -132,23 +132,23 @@ private LocalLogManagerTestEnv( dir = TestUtils.tempDirectory(); shared = new SharedLogData(snapshotReader); sharedLogDataInitializer.accept(shared); - List newLogManagers = new ArrayList<>(numManagers); + List newRaftClients = new ArrayList<>(numNodes); try { - for (int nodeId = 0; nodeId < numManagers; nodeId++) { - newLogManagers.add(new LocalLogManager( - new LogContext(String.format("[LocalLogManager %d] ", nodeId)), + for (int nodeId = 0; nodeId < numNodes; nodeId++) { + newRaftClients.add(new MockRaftClient( + new LogContext(String.format("[MockRaftClient %d] ", nodeId)), nodeId, shared, - String.format("LocalLogManager-%d_", nodeId), + String.format("MockRaftClient-%d_", nodeId), lastKRaftVersion)); } } catch (Throwable t) { - for (LocalLogManager logManager : newLogManagers) { - logManager.close(); + for (MockRaftClient raftClient : newRaftClients) { + raftClient.close(); } throw t; } - this.logManagers = newLogManagers; + this.raftClients = newRaftClients; } /** @@ -179,9 +179,9 @@ LeaderAndEpoch waitForLeader() throws InterruptedException { AtomicReference value = new AtomicReference<>(null); TestUtils.retryOnExceptionWithTimeout(20000, 3, () -> { LeaderAndEpoch result = null; - for (LocalLogManager logManager : logManagers) { - LeaderAndEpoch leader = logManager.leaderAndEpoch(); - int nodeId = logManager.nodeId().getAsInt(); + for (MockRaftClient raftClient : raftClients) { + LeaderAndEpoch leader = raftClient.leaderAndEpoch(); + int nodeId = raftClient.nodeId().getAsInt(); if (leader.isLeader(nodeId)) { if (result != null) { throw new RuntimeException("node " + nodeId + @@ -198,14 +198,14 @@ LeaderAndEpoch waitForLeader() throws InterruptedException { return value.get(); } - public List logManagers() { - return logManagers; + public List raftClients() { + return raftClients; } - public Optional activeLogManager() { + public Optional activeRaftClient() { OptionalInt leader = shared.leaderAndEpoch().leaderId(); if (leader.isPresent()) { - return Optional.of(logManagers.get(leader.getAsInt())); + return Optional.of(raftClients.get(leader.getAsInt())); } else { return Optional.empty(); } @@ -218,11 +218,11 @@ public LeaderAndEpoch leaderAndEpoch() { @Override public void close() throws InterruptedException { try { - for (LocalLogManager logManager : logManagers) { - logManager.beginShutdown(); + for (MockRaftClient raftClient : raftClients) { + raftClient.beginShutdown(); } - for (LocalLogManager logManager : logManagers) { - logManager.close(); + for (MockRaftClient raftClient : raftClients) { + raftClient.close(); } Utils.delete(dir); } catch (IOException e) { diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java index 7e082ac303326..894a7f739c468 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerMetricsIntegrationTest.java @@ -24,7 +24,6 @@ import org.apache.kafka.common.utils.Time; import org.apache.kafka.controller.metrics.QuorumControllerMetrics; import org.apache.kafka.metadata.BrokerHeartbeatReply; -import org.apache.kafka.metalog.LocalLogManagerTestEnv; import org.apache.kafka.test.TestUtils; import org.junit.jupiter.api.Test; @@ -74,9 +73,9 @@ public void close() { public void testClosingQuorumControllerClosesMetrics() throws Throwable { MockControllerMetrics metrics = new MockControllerMetrics(); try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setControllerBuilderInitializer(controllerBuilder -> controllerBuilder.setMetrics(metrics) ). @@ -97,9 +96,9 @@ public void testFailingOverIncrementsNewActiveControllerCount( boolean forceFailoverUsingLogLayer ) throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { registerBrokersAndUnfence(controlEnv.activeController(), 1); // wait for a controller to become active. @@ -109,7 +108,7 @@ public void testFailingOverIncrementsNewActiveControllerCount( } }); if (forceFailoverUsingLogLayer) { - logEnv.activeLogManager().get().throwOnNextAppend(); + clientEnv.activeRaftClient().get().throwOnNextAppend(); TestUtils.retryOnExceptionWithTimeout(30_000, () -> createTopics(controlEnv.activeController(), "test_", 1, 1) @@ -133,9 +132,9 @@ public void testFailingOverIncrementsNewActiveControllerCount( @Test public void testTimeoutMetrics() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { QuorumController active = controlEnv.activeController(); @@ -192,9 +191,9 @@ public void testTimeoutMetrics() throws Throwable { @Test public void testEventQueueOperationsStartedMetric() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { QuorumController active = controlEnv.activeController(); diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java index 41ef67ab885ae..3e58cabeac311 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java @@ -99,8 +99,6 @@ import org.apache.kafka.metadata.RecordTestUtils.TestThroughAllIntermediateImagesLeadingToFinalImageHelper; import org.apache.kafka.metadata.bootstrap.BootstrapMetadata; import org.apache.kafka.metadata.util.BatchFileWriter; -import org.apache.kafka.metalog.LocalLogManager; -import org.apache.kafka.metalog.LocalLogManagerTestEnv; import org.apache.kafka.raft.Batch; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.apache.kafka.server.common.EligibleLeaderReplicasVersion; @@ -180,9 +178,9 @@ public class QuorumControllerTest { @Test public void testConfigurationOperations() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT, @@ -191,10 +189,10 @@ public void testConfigurationOperations() throws Throwable { Map.of(EligibleLeaderReplicasVersion.FEATURE_NAME, EligibleLeaderReplicasVersion.ELRV_1.featureLevel()))). setBrokerId(0). setLogDirs(List.of(Uuid.fromString("iiaQjkRPQcuMULNII0MUeA"))). - setClusterId(logEnv.clusterId())).get(); + setClusterId(clientEnv.clusterId())).get(); testConfigurationOperations(controlEnv.activeController()); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @@ -221,9 +219,9 @@ private void testConfigurationOperations(QuorumController controller) throws Thr @Test public void testDelayedConfigurationOperations() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { controlEnv.activeController().registerBroker(ANONYMOUS_CONTEXT, @@ -232,18 +230,18 @@ public void testDelayedConfigurationOperations() throws Throwable { Map.of(EligibleLeaderReplicasVersion.FEATURE_NAME, EligibleLeaderReplicasVersion.ELRV_1.featureLevel()))). setBrokerId(0). setLogDirs(List.of(Uuid.fromString("sTbzRAMnTpahIyIPNjiLhw"))). - setClusterId(logEnv.clusterId())).get(); - testDelayedConfigurationOperations(logEnv, controlEnv.activeController()); + setClusterId(clientEnv.clusterId())).get(); + testDelayedConfigurationOperations(clientEnv, controlEnv.activeController()); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } private void testDelayedConfigurationOperations( - LocalLogManagerTestEnv logEnv, + MockRaftClientTestEnv clientEnv, QuorumController controller ) throws Throwable { - logEnv.logManagers().forEach(m -> m.setMaxReadOffset(1L)); + clientEnv.raftClients().forEach(m -> m.setMaxReadOffset(1L)); CompletableFuture> future1 = controller.incrementalAlterConfigs(ANONYMOUS_CONTEXT, Map.of( BROKER0, Map.of("baz", entry(SET, "123"))), false); @@ -252,7 +250,7 @@ private void testDelayedConfigurationOperations( new ResultOrError<>(Map.of())), controller.describeConfigs(ANONYMOUS_CONTEXT, Map.of( BROKER0, List.of())).get()); - logEnv.logManagers().forEach(m -> m.setMaxReadOffset(8L)); + clientEnv.raftClients().forEach(m -> m.setMaxReadOffset(8L)); assertEquals(Map.of(BROKER0, ApiError.NONE), future1.get()); } @@ -266,9 +264,9 @@ public void testFenceMultipleBrokers() throws Throwable { long sessionTimeoutMillis = 1000; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)). setBootstrapMetadata(SIMPLE_BOOTSTRAP). build() @@ -344,7 +342,7 @@ public void testFenceMultipleBrokers() throws Throwable { // Check that there are imbalaned partitions assertTrue(active.replicationControl().arePartitionLeadersImbalanced()); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @@ -352,9 +350,9 @@ public void testFenceMultipleBrokers() throws Throwable { public void testElrEnabledByDefault() throws Throwable { long sessionTimeoutMillis = 500; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)). setBootstrapMetadata(BootstrapMetadata.fromRecords( List.of( @@ -384,9 +382,9 @@ public void testUncleanShutdownBrokerElrEnabled() throws Throwable { long sessionTimeoutMillis = 500; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)). setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_4_0_IV1, "test-provided bootstrap ELR enabled")). build() @@ -521,9 +519,9 @@ public void testUncleanShutdownElrDisabled() throws Exception { long sessionTimeoutMillis = 500; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv) + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv) .setControllerBuilderInitializer(controllerBuilder -> controllerBuilder.setFenceStaleBrokerIntervalNs(TimeUnit.SECONDS.toNanos(15))) .setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)) @@ -618,8 +616,8 @@ public void testMinIsrUpdateWithElr() throws Throwable { long sessionTimeoutMillis = 300; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1).build(); + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)). setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_4_0_IV1, "test-provided bootstrap ELR enabled")). build() @@ -747,9 +745,9 @@ public void testBalancePartitionLeaders() throws Throwable { long leaderImbalanceCheckIntervalNs = 1_000_000_000; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setSessionTimeoutMillis(OptionalLong.of(sessionTimeoutMillis)). setLeaderImbalanceCheckIntervalNs(OptionalLong.of(leaderImbalanceCheckIntervalNs)). setBootstrapMetadata(SIMPLE_BOOTSTRAP). @@ -873,7 +871,7 @@ public void testBalancePartitionLeaders() throws Throwable { "Leaders were not balanced after unfencing all of the brokers" ); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @@ -886,9 +884,9 @@ public void testNoOpRecordWriteAfterTimeout() throws Throwable { long maxIdleIntervalNs = TimeUnit.MICROSECONDS.toNanos(100); long maxReplicationDelayMs = 1_000; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setControllerBuilderInitializer(controllerBuilder -> controllerBuilder.setMaxIdleIntervalNs(OptionalLong.of(maxIdleIntervalNs)) ). @@ -898,29 +896,29 @@ public void testNoOpRecordWriteAfterTimeout() throws Throwable { listeners.add(new Listener().setName("PLAINTEXT").setHost("localhost").setPort(9092)); QuorumController active = controlEnv.activeController(); - LocalLogManager localLogManager = logEnv - .logManagers() + MockRaftClient mockRaftClient = clientEnv + .raftClients() .stream() .filter(logManager -> logManager.nodeId().equals(OptionalInt.of(active.nodeId()))) .findAny() .get(); TestUtils.waitForCondition( - () -> localLogManager.highWatermark().isPresent(), + () -> mockRaftClient.highWatermark().isPresent(), maxReplicationDelayMs, "High watermark was not established" ); - final long firstHighWatermark = localLogManager.highWatermark().getAsLong(); + final long firstHighWatermark = mockRaftClient.highWatermark().getAsLong(); TestUtils.waitForCondition( - () -> localLogManager.highWatermark().getAsLong() > firstHighWatermark, + () -> mockRaftClient.highWatermark().getAsLong() > firstHighWatermark, maxReplicationDelayMs, "Active controller didn't write NoOpRecord the first time" ); // Do it again to make sure that we are not counting the leader change record - final long secondHighWatermark = localLogManager.highWatermark().getAsLong(); + final long secondHighWatermark = mockRaftClient.highWatermark().getAsLong(); TestUtils.waitForCondition( - () -> localLogManager.highWatermark().getAsLong() > secondHighWatermark, + () -> mockRaftClient.highWatermark().getAsLong() > secondHighWatermark, maxReplicationDelayMs, "Active controller didn't write NoOpRecord the second time" ); @@ -931,10 +929,10 @@ public void testNoOpRecordWriteAfterTimeout() throws Throwable { @CsvSource(value = {"0, 0", "0, 1", "1, 0", "1, 1"}) public void testRegisterBrokerKRaftVersions(short finalizedKraftVersion, short brokerMaxSupportedKraftVersion) throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). setLastKRaftVersion(KRaftVersion.fromFeatureLevel(finalizedKraftVersion)). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setBootstrapMetadata(SIMPLE_BOOTSTRAP). build() ) { @@ -983,9 +981,9 @@ public void testRegisterBrokerKRaftVersions(short finalizedKraftVersion, short b @Test public void testUnregisterBroker() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { ListenerCollection listeners = new ListenerCollection(); @@ -1041,7 +1039,7 @@ public void testUnregisterBroker() throws Throwable { }); assertEquals(0, topicPartitionFuture.get().partitionId()); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @@ -1063,14 +1061,14 @@ public void testSnapshotSaveAndLoad() throws Throwable { Map brokerEpochs = new HashMap<>(); Uuid fooId; try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setBootstrapMetadata(SIMPLE_BOOTSTRAP). build() ) { QuorumController active = controlEnv.activeController(); - for (int i = 0; i < logEnv.logManagers().size(); i++) { + for (int i = 0; i < clientEnv.raftClients().size(); i++) { active.registerController(ANONYMOUS_CONTEXT, new ControllerRegistrationRequestData(). setControllerId(i). @@ -1131,9 +1129,9 @@ public void testSnapshotSaveAndLoad() throws Throwable { active.allocateProducerIds(ANONYMOUS_CONTEXT, new AllocateProducerIdsRequestData().setBrokerId(0).setBrokerEpoch(brokerEpochs.get(0))).get(); controlEnv.close(); - assertEquals(generateTestRecords(fooId, brokerEpochs), logEnv.allRecords()); + assertEquals(generateTestRecords(fooId, brokerEpochs), clientEnv.allRecords()); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @@ -1269,8 +1267,8 @@ private List generateTestRecords(Uuid fooId, Map future) { @Test public void testEarlyControllerResults() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { QuorumController controller = controlEnv.activeController(); @@ -1362,16 +1360,16 @@ public void testEarlyControllerResults() throws Throwable { alterReassignmentsFuture.get(); countDownLatch.countDown(); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @Test public void testConfigResourceExistenceChecker() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { QuorumController active = controlEnv.activeController(); @@ -1402,16 +1400,16 @@ public void testConfigResourceExistenceChecker() throws Throwable { assertThrows(UnknownTopicOrPartitionException.class, () -> checker.accept(new ConfigResource(TOPIC, "bar"))); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } @Test public void testFatalMetadataReplayErrorOnActive() throws Throwable { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). build() ) { QuorumController active = controlEnv.activeController(); @@ -1435,14 +1433,14 @@ public void testFatalMetadataErrorDuringSnapshotLoading() throws Exception { new ApiMessageAndVersion(new PartitionRecord(), (short) 0)) ); - LocalLogManagerTestEnv.Builder logEnvBuilder = new LocalLogManagerTestEnv.Builder(3) + MockRaftClientTestEnv.Builder clientEnvBuilder = new MockRaftClientTestEnv.Builder(3) .setSnapshotReader(FileRawSnapshotReader.open( invalidSnapshot.tempDir.toPath(), new OffsetAndEpoch(0, 0) )); - try (LocalLogManagerTestEnv logEnv = logEnvBuilder.build()) { - try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).build()) { + try (MockRaftClientTestEnv clientEnv = clientEnvBuilder.build()) { + try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv).build()) { TestUtils.waitForCondition(() -> controlEnv.controllers().stream().allMatch( controller -> controlEnv.fatalFaultHandler(controller.nodeId()).firstException() != null), "At least one controller failed to detect the fatal fault" @@ -1454,11 +1452,11 @@ public void testFatalMetadataErrorDuringSnapshotLoading() throws Exception { @Test public void testFatalMetadataErrorDuringLogLoading() throws Exception { - try (LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3).build()) { - logEnv.appendInitialRecords(List.of( + try (MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3).build()) { + clientEnv.appendInitialRecords(List.of( new ApiMessageAndVersion(new PartitionRecord(), (short) 0))); - try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).build()) { + try (QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv).build()) { TestUtils.waitForCondition(() -> controlEnv.controllers().stream().allMatch( controller -> controlEnv.fatalFaultHandler(controller.nodeId()).firstException() != null), "At least one controller failed to detect the fatal fault" @@ -1505,9 +1503,9 @@ public void close() throws Exception { @Test public void testInsertBootstrapRecordsToEmptyLog() throws Exception { try ( - LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3). + MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(3). build(); - QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv). + QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv). setBootstrapMetadata(COMPLEX_BOOTSTRAP). build() ) { @@ -1538,7 +1536,7 @@ public void testInsertBootstrapRecordsToEmptyLog() throws Exception { Map.of("foo", "bar").equals(resultOrError.result()); }, "Failed to see expected config change from bootstrap metadata"); - testToImages(logEnv.allRecords()); + testToImages(clientEnv.allRecords()); } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java index b04a3e8c65874..e6ececd2ba19c 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java +++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java @@ -19,7 +19,6 @@ import org.apache.kafka.metadata.FakeKafkaConfigSchema; import org.apache.kafka.metadata.bootstrap.BootstrapMetadata; -import org.apache.kafka.metalog.LocalLogManagerTestEnv; import org.apache.kafka.raft.LeaderAndEpoch; import org.apache.kafka.server.common.EligibleLeaderReplicasVersion; import org.apache.kafka.server.common.MetadataVersion; @@ -41,20 +40,20 @@ public class QuorumControllerTestEnv implements AutoCloseable { private final List controllers; - private final LocalLogManagerTestEnv logEnv; + private final MockRaftClientTestEnv clientEnv; private final Map fatalFaultHandlers = new HashMap<>(); private final Map nonFatalFaultHandlers = new HashMap<>(); public static class Builder { - private final LocalLogManagerTestEnv logEnv; + private final MockRaftClientTestEnv clientEnv; private Consumer controllerBuilderInitializer = __ -> { }; private OptionalLong sessionTimeoutMillis = OptionalLong.empty(); private OptionalLong leaderImbalanceCheckIntervalNs = OptionalLong.empty(); private BootstrapMetadata bootstrapMetadata = BootstrapMetadata. fromVersion(MetadataVersion.latestTesting(), "test-provided version"); - public Builder(LocalLogManagerTestEnv logEnv) { - this.logEnv = logEnv; + public Builder(MockRaftClientTestEnv clientEnv) { + this.clientEnv = clientEnv; } public Builder setControllerBuilderInitializer(Consumer controllerBuilderInitializer) { @@ -79,7 +78,7 @@ public Builder setBootstrapMetadata(BootstrapMetadata bootstrapMetadata) { public QuorumControllerTestEnv build() throws Exception { return new QuorumControllerTestEnv( - logEnv, + clientEnv, controllerBuilderInitializer, sessionTimeoutMillis, leaderImbalanceCheckIntervalNs, @@ -89,21 +88,21 @@ public QuorumControllerTestEnv build() throws Exception { } private QuorumControllerTestEnv( - LocalLogManagerTestEnv logEnv, + MockRaftClientTestEnv clientEnv, Consumer controllerBuilderInitializer, OptionalLong sessionTimeoutMillis, OptionalLong leaderImbalanceCheckIntervalNs, boolean eligibleLeaderReplicasEnabled, BootstrapMetadata bootstrapMetadata ) throws Exception { - this.logEnv = logEnv; - int numControllers = logEnv.logManagers().size(); + this.clientEnv = clientEnv; + int numControllers = clientEnv.raftClients().size(); this.controllers = new ArrayList<>(numControllers); try { List nodeIds = IntStream.range(0, numControllers).boxed().toList(); for (int nodeId = 0; nodeId < numControllers; nodeId++) { - QuorumController.Builder builder = new QuorumController.Builder(nodeId, logEnv.clusterId()); - builder.setRaftClient(logEnv.logManagers().get(nodeId)); + QuorumController.Builder builder = new QuorumController.Builder(nodeId, clientEnv.clusterId()); + builder.setRaftClient(clientEnv.raftClients().get(nodeId)); if (eligibleLeaderReplicasEnabled) { bootstrapMetadata = bootstrapMetadata.copyWithFeatureRecord( EligibleLeaderReplicasVersion.FEATURE_NAME, @@ -140,7 +139,7 @@ QuorumController activeController() throws InterruptedException { QuorumController activeController(boolean waitForActivation) throws InterruptedException { AtomicReference value = new AtomicReference<>(null); TestUtils.retryOnExceptionWithTimeout(20000, 3, () -> { - LeaderAndEpoch leader = logEnv.leaderAndEpoch(); + LeaderAndEpoch leader = clientEnv.leaderAndEpoch(); for (QuorumController controller : controllers) { if (OptionalInt.of(controller.nodeId()).equals(leader.leaderId()) && controller.curClaimEpoch() == leader.epoch()) {