apache · chia7712 · Jun 19, 2024 · Jun 19, 2024
diff --git a/build.gradle b/build.gradle
@@ -203,7 +203,6 @@ def determineCommitId() {
 
 def excludedSpotlessModules = [':clients',
                                ':connect:runtime',
-                               ':core',
                                ':streams']
 
 

diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -16,8 +16,6 @@
  */
 package kafka.log.remote;
 
-import com.yammer.metrics.core.Gauge;
-import com.yammer.metrics.core.Timer;
 import kafka.cluster.EndPoint;
 import kafka.cluster.Partition;
 import kafka.log.UnifiedLog;
@@ -27,6 +25,7 @@
 import kafka.server.KafkaConfig;
 import kafka.server.QuotaType;
 import kafka.server.StopPartition;
+
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
@@ -82,10 +81,11 @@
 import org.apache.kafka.storage.internals.log.TransactionIndex;
 import org.apache.kafka.storage.internals.log.TxnIndexSearchResult;
 
+import com.yammer.metrics.core.Gauge;
+import com.yammer.metrics.core.Timer;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.Option;
-import scala.collection.JavaConverters;
 
 import java.io.BufferedWriter;
 import java.io.ByteArrayOutputStream;
@@ -108,8 +108,8 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.NavigableMap;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.OptionalLong;
@@ -135,10 +135,13 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import scala.Option;
+import scala.collection.JavaConverters;
+
+import static org.apache.kafka.server.config.ServerLogConfigs.LOG_DIR_CONFIG;
 import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX;
 import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC;
 import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_READER_FETCH_RATE_AND_TIME_METRIC;
-import static org.apache.kafka.server.config.ServerLogConfigs.LOG_DIR_CONFIG;
 
 /**
  * This class is responsible for

diff --git a/core/src/main/java/kafka/log/remote/RemoteLogReader.java b/core/src/main/java/kafka/log/remote/RemoteLogReader.java
@@ -16,14 +16,17 @@
  */
 package kafka.log.remote;
 
-import com.yammer.metrics.core.Timer;
 import kafka.log.remote.quota.RLMQuotaManager;
 import kafka.server.BrokerTopicStats;
+
 import org.apache.kafka.common.errors.OffsetOutOfRangeException;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.storage.internals.log.FetchDataInfo;
 import org.apache.kafka.storage.internals.log.RemoteLogReadResult;
 import org.apache.kafka.storage.internals.log.RemoteStorageFetchInfo;
+
+import com.yammer.metrics.core.Timer;
+
 import org.slf4j.Logger;
 
 import java.util.Optional;

diff --git a/core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java b/core/src/main/java/kafka/log/remote/quota/RLMQuotaManager.java
@@ -18,6 +18,7 @@
 
 import kafka.server.QuotaType;
 import kafka.server.SensorAccess;
+
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.MetricConfig;
@@ -27,15 +28,17 @@
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.metrics.stats.SimpleRate;
 import org.apache.kafka.common.utils.Time;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.runtime.BoxedUnit;
 
 import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 
+import scala.runtime.BoxedUnit;
+
 public class RLMQuotaManager {
     private static final Logger LOGGER = LoggerFactory.getLogger(RLMQuotaManager.class);
 

diff --git a/core/src/main/java/kafka/server/TierStateMachine.java b/core/src/main/java/kafka/server/TierStateMachine.java
@@ -17,23 +17,10 @@
 
 package kafka.server;
 
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.nio.file.StandardCopyOption;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
 import kafka.cluster.Partition;
 import kafka.log.UnifiedLog;
 import kafka.log.remote.RemoteLogManager;
+
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.message.FetchResponseData.PartitionData;
@@ -49,8 +36,24 @@
 import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile;
 import org.apache.kafka.storage.internals.log.EpochEntry;
 import org.apache.kafka.storage.internals.log.LogFileUtils;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import scala.Option;
 import scala.collection.JavaConverters;
 

diff --git a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
@@ -31,6 +31,7 @@
 import kafka.server.QuotaFactory.QuotaManagers;
 import kafka.server.ReplicaManager;
 import kafka.server.metadata.ConfigRepository;
+
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.coordinator.group.GroupCoordinator;

diff --git a/core/src/main/java/kafka/server/builders/LogManagerBuilder.java b/core/src/main/java/kafka/server/builders/LogManagerBuilder.java
@@ -20,20 +20,22 @@
 import kafka.log.LogManager;
 import kafka.server.BrokerTopicStats;
 import kafka.server.metadata.ConfigRepository;
+
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.server.config.ServerLogConfigs;
+import org.apache.kafka.server.util.Scheduler;
 import org.apache.kafka.storage.internals.log.CleanerConfig;
 import org.apache.kafka.storage.internals.log.LogConfig;
 import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
-import org.apache.kafka.server.util.Scheduler;
 import org.apache.kafka.storage.internals.log.ProducerStateManagerConfig;
-import scala.collection.JavaConverters;
 
 import java.io.File;
 import java.util.Collections;
 import java.util.List;
 
+import scala.collection.JavaConverters;
+
 
 public class LogManagerBuilder {
     private static final int PRODUCER_ID_EXPIRATION_CHECK_INTERVAL_MS = 600000;

diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -18,6 +18,7 @@
 package kafka.server.builders;
 
 import kafka.log.LogManager;
+import kafka.log.remote.RemoteLogManager;
 import kafka.server.AddPartitionsToTxnManager;
 import kafka.server.AlterPartitionManager;
 import kafka.server.BrokerTopicStats;
@@ -31,19 +32,20 @@
 import kafka.server.MetadataCache;
 import kafka.server.QuotaFactory.QuotaManagers;
 import kafka.server.ReplicaManager;
-import kafka.log.remote.RemoteLogManager;
 import kafka.zk.KafkaZkClient;
+
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.server.common.DirectoryEventHandler;
-import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
 import org.apache.kafka.server.util.Scheduler;
-import scala.compat.java8.OptionConverters;
+import org.apache.kafka.storage.internals.log.LogDirFailureChannel;
 
 import java.util.Collections;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import scala.compat.java8.OptionConverters;
+
 
 public class ReplicaManagerBuilder {
     private KafkaConfig config = null;

diff --git a/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java b/core/src/main/java/kafka/server/handlers/DescribeTopicPartitionsRequestHandler.java
@@ -21,6 +21,7 @@
 import kafka.server.AuthHelper;
 import kafka.server.KafkaConfig;
 import kafka.server.metadata.KRaftMetadataCache;
+
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.message.DescribeTopicPartitionsRequestData;
@@ -30,14 +31,15 @@
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.DescribeTopicPartitionsRequest;
 import org.apache.kafka.common.resource.Resource;
-import scala.collection.JavaConverters;
 
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.stream.Stream;
 
+import scala.collection.JavaConverters;
+
 import static org.apache.kafka.common.acl.AclOperation.DESCRIBE;
 import static org.apache.kafka.common.resource.ResourceType.TOPIC;
 

diff --git a/core/src/main/java/kafka/server/logger/RuntimeLoggerManager.java b/core/src/main/java/kafka/server/logger/RuntimeLoggerManager.java
@@ -18,6 +18,7 @@
 package kafka.server.logger;
 
 import kafka.utils.Log4jController;
+
 import org.apache.kafka.clients.admin.AlterConfigOp.OpType;
 import org.apache.kafka.common.config.LogLevelConfig;
 import org.apache.kafka.common.errors.ClusterAuthorizationException;
@@ -26,6 +27,7 @@
 import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterConfigsResource;
 import org.apache.kafka.common.message.IncrementalAlterConfigsRequestData.AlterableConfig;
 import org.apache.kafka.common.protocol.Errors;
+
 import org.slf4j.Logger;
 
 import java.util.ArrayList;

diff --git a/core/src/main/java/kafka/server/share/ErroneousAndValidPartitionData.java b/core/src/main/java/kafka/server/share/ErroneousAndValidPartitionData.java
@@ -22,12 +22,13 @@
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.ShareFetchRequest;
 import org.apache.kafka.common.requests.ShareFetchResponse;
-import scala.Tuple2;
 
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 
+import scala.Tuple2;
+
 /**
  * Helper class to return the erroneous partitions and valid partition data
  */

diff --git a/core/src/main/java/kafka/server/share/FinalContext.java b/core/src/main/java/kafka/server/share/FinalContext.java
@@ -22,6 +22,7 @@
 import org.apache.kafka.common.message.ShareFetchResponseData;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.ShareFetchResponse;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/core/src/main/java/kafka/server/share/ShareFetchContext.java b/core/src/main/java/kafka/server/share/ShareFetchContext.java
@@ -17,6 +17,7 @@
 package kafka.server.share;
 
 import kafka.server.FetchSession;
+
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.ShareFetchResponseData;

diff --git a/core/src/main/java/kafka/server/share/SharePartition.java b/core/src/main/java/kafka/server/share/SharePartition.java
@@ -17,6 +17,7 @@
 package kafka.server.share;
 
 import kafka.server.ReplicaManager;
+
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.InvalidRecordStateException;
@@ -38,6 +39,7 @@
 import org.apache.kafka.server.util.timer.Timer;
 import org.apache.kafka.server.util.timer.TimerTask;
 import org.apache.kafka.storage.internals.log.FetchPartitionData;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/core/src/main/java/kafka/server/share/SharePartitionManager.java b/core/src/main/java/kafka/server/share/SharePartitionManager.java
@@ -18,6 +18,7 @@
 
 import kafka.server.FetchSession;
 import kafka.server.ReplicaManager;
+
 import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.Uuid;
@@ -31,11 +32,12 @@
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.server.group.share.Persister;
 import org.apache.kafka.server.share.CachedSharePartition;
+import org.apache.kafka.server.share.ShareAcknowledgementBatch;
 import org.apache.kafka.server.share.ShareSession;
 import org.apache.kafka.server.share.ShareSessionCache;
 import org.apache.kafka.server.share.ShareSessionKey;
-import org.apache.kafka.server.share.ShareAcknowledgementBatch;
 import org.apache.kafka.storage.internals.log.FetchParams;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/core/src/main/java/kafka/server/share/ShareSessionContext.java b/core/src/main/java/kafka/server/share/ShareSessionContext.java
@@ -29,9 +29,9 @@
 import org.apache.kafka.common.requests.ShareFetchResponse;
 import org.apache.kafka.server.share.CachedSharePartition;
 import org.apache.kafka.server.share.ShareSession;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import scala.Tuple2;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -42,6 +42,8 @@
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
 
+import scala.Tuple2;
+
 /**
  * The context for a share session fetch request.
  */

diff --git a/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java b/core/src/test/java/kafka/admin/ConfigCommandIntegrationTest.java
@@ -26,6 +26,7 @@
 import kafka.zk.AdminZkClient;
 import kafka.zk.BrokerInfo;
 import kafka.zk.KafkaZkClient;
+
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.ConfigResource;
@@ -36,6 +37,7 @@
 import org.apache.kafka.server.common.MetadataVersion;
 import org.apache.kafka.server.config.ZooKeeperInternals;
 import org.apache.kafka.test.TestUtils;
+
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.platform.commons.util.StringUtils;