Skip to content

MINOR: Remove the MetaLogShim namings #20357

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 3 commits into from
Aug 15, 2025
Merged

MINOR: Remove the MetaLogShim namings #20357

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3b357dc
MINOR: Fix setRaftClient parameter name error
mapan1984 Aug 14, 2025
e44602c
MINOR: Remove the MetaLogShim namings
mapan1984 Aug 15, 2025
5dd271e
MINOR: Remove metalog checkstyle
mapan1984 Aug 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 0 additions & 16 deletions checkstyle/import-control-metadata.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,6 @@
<allow pkg="org.apache.kafka.metadata" />
<allow pkg="org.apache.kafka.metadata.authorizer" />
<allow pkg="org.apache.kafka.metadata.migration" />
<allow pkg="org.apache.kafka.metalog" />
<allow pkg="org.apache.kafka.deferred" />
<allow pkg="org.apache.kafka.queue" />
<allow pkg="org.apache.kafka.raft" />
Expand Down Expand Up @@ -160,7 +159,6 @@
<allow pkg="org.apache.kafka.common.requests" />
<allow pkg="org.apache.kafka.image" />
<allow pkg="org.apache.kafka.metadata" />
<allow pkg="org.apache.kafka.metalog" />
<allow pkg="org.apache.kafka.queue" />
<allow pkg="org.apache.kafka.raft" />
<allow pkg="org.apache.kafka.server.authorizer" />
Expand Down Expand Up @@ -198,18 +196,4 @@
</subpackage>
</subpackage>

<subpackage name="metalog">
<allow class="org.apache.kafka.common.compress.Compression" exact-match="true" />
<allow pkg="org.apache.kafka.common.metadata" />
<allow pkg="org.apache.kafka.common.protocol" />
<allow pkg="org.apache.kafka.common.record" />
<allow pkg="org.apache.kafka.metadata" />
<allow pkg="org.apache.kafka.metalog" />
<allow pkg="org.apache.kafka.raft" />
<allow pkg="org.apache.kafka.snapshot" />
<allow pkg="org.apache.kafka.queue" />
<allow pkg="org.apache.kafka.server.common" />
<allow pkg="org.apache.kafka.test" />
</subpackage>

</import-control>
6 changes: 3 additions & 3 deletions metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,8 +260,8 @@ public Builder setConfigSchema(KafkaConfigSchema configSchema) {
return this;
}

public Builder setRaftClient(RaftClient<ApiMessageAndVersion> logManager) {
this.raftClient = logManager;
public Builder setRaftClient(RaftClient<ApiMessageAndVersion> raftClient) {
this.raftClient = raftClient;
return this;
}

Expand Down Expand Up @@ -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) {
Expand Down
82 changes: 41 additions & 41 deletions ...apache/kafka/metalog/LocalLogManager.java → ...ache/kafka/controller/MockRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
* limitations under the License.
*/

package org.apache.kafka.metalog;

@chia7712 chia7712 Aug 15, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could you please remove org.apache.kafka.metalog from import-control-metadata.xml?

@mapan1984 mapan1984 Aug 15, 2025

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, have been removed.

package org.apache.kafka.controller;

import org.apache.kafka.common.protocol.ObjectSerializationCache;
import org.apache.kafka.common.utils.BufferSupplier;
Expand Down Expand Up @@ -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<ApiMessageAndVersion>, AutoCloseable {
public final class MockRaftClient implements RaftClient<ApiMessageAndVersion>, AutoCloseable {
interface LocalBatch {
int epoch();
int size();
Expand Down Expand Up @@ -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<Integer, LocalLogManager> logManagers = new HashMap<>();
private final HashMap<Integer, MockRaftClient> raftClients = new HashMap<>();

/**
* Maps offsets to record batches.
Expand Down Expand Up @@ -198,17 +198,17 @@ public SharedLogData(Optional<RawSnapshotReader> 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.");
}
}
Expand Down Expand Up @@ -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<Integer> iter = logManagers.keySet().iterator();
int nextLeaderIndex = ThreadLocalRandom.current().nextInt(raftClients.size());
Iterator<Integer> iter = raftClients.keySet().iterator();
Integer nextLeaderNode = null;
for (int i = 0; i <= nextLeaderIndex; i++) {
nextLeaderNode = iter.next();
Expand All @@ -294,7 +294,7 @@ synchronized Entry<Long, LocalBatch> 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<RawSnapshotReader> nextSnapshot(long offset) {
return Optional.ofNullable(snapshots.lastEntry()).flatMap(entry -> {
Expand Down Expand Up @@ -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<OffsetAndEpoch> latestSnapshotId() {
return Optional.ofNullable(snapshots.lastEntry()).map(entry -> entry.getValue().snapshotId());
Expand Down Expand Up @@ -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<Listener<ApiMessageAndVersion>, 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() {
Expand All @@ -477,7 +477,7 @@ private void scheduleLogCheck() {
listenerData.handleLoadSnapshot(
RecordsSnapshotReader.of(
snapshot.get(),
new MetadataRecordSerde(),
new MetadataRecordSerde(),
BufferSupplier.create(),
Integer.MAX_VALUE,
true,
Expand Down Expand Up @@ -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 {
Expand All @@ -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);
Expand All @@ -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.
Expand All @@ -614,7 +614,7 @@ public void register(RaftClient.Listener<ApiMessageAndVersion> listener) {
CompletableFuture<Void> 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 {
Expand Down Expand Up @@ -643,7 +643,7 @@ public void register(RaftClient.Listener<ApiMessageAndVersion> listener) {
public void unregister(RaftClient.Listener<ApiMessageAndVersion> 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) {
Expand Down
6 changes: 3 additions & 3 deletions ...a/metalog/MockMetaLogManagerListener.java → ...ka/controller/MockRaftClientListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -29,7 +29,7 @@
import java.util.List;
import java.util.OptionalInt;

public class MockMetaLogManagerListener implements RaftClient.Listener<ApiMessageAndVersion> {
public class MockRaftClientListener implements RaftClient.Listener<ApiMessageAndVersion> {
public static final String COMMIT = "COMMIT";
public static final String LAST_COMMITTED_OFFSET = "LAST_COMMITTED_OFFSET";
public static final String NEW_LEADER = "NEW_LEADER";
Expand All @@ -41,7 +41,7 @@ public class MockMetaLogManagerListener implements RaftClient.Listener<ApiMessag
private final List<String> serializedEvents = new ArrayList<>();
private LeaderAndEpoch leaderAndEpoch = new LeaderAndEpoch(OptionalInt.empty(), 0);

public MockMetaLogManagerListener(int nodeId) {
public MockRaftClientListener(int nodeId) {
this.nodeId = nodeId;
}

Expand Down
Loading
Loading