KAFKA-17185: Declare Loggers as static to prevent multiple logger instances

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

@@ -45,7 +45,7 @@
  */
 public class AbstractConfig {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(AbstractConfig.class);
 
     /**
      * Configs for which values have been requested, used to detect unused configs.

clients/src/main/java/org/apache/kafka/common/metrics/stats/Percentiles.java

@@ -33,7 +33,7 @@
  */
 public class Percentiles extends SampledStat implements CompoundStat {
 
-    private final Logger log = LoggerFactory.getLogger(Percentiles.class);
+    private static final Logger log = LoggerFactory.getLogger(Percentiles.class);
 
     public enum BucketSizing {
         CONSTANT, LINEAR

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/unsecured/OAuthBearerUnsecuredLoginCallbackHandler.java

@@ -99,7 +99,7 @@
  * broker configuration property.
  */
 public class OAuthBearerUnsecuredLoginCallbackHandler implements AuthenticateCallbackHandler {
-    private final Logger log = LoggerFactory.getLogger(OAuthBearerUnsecuredLoginCallbackHandler.class);
+    private static final Logger log = LoggerFactory.getLogger(OAuthBearerUnsecuredLoginCallbackHandler.class);
     private static final String OPTION_PREFIX = "unsecuredLogin";
     private static final String PRINCIPAL_CLAIM_NAME_OPTION = OPTION_PREFIX + "PrincipalClaimName";
     private static final String LIFETIME_SECONDS_OPTION = OPTION_PREFIX + "LifetimeSeconds";

clients/src/main/java/org/apache/kafka/common/utils/KafkaThread.java

@@ -24,7 +24,7 @@
  */
 public class KafkaThread extends Thread {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(KafkaThread.class);
 
     public static KafkaThread daemon(final String name, Runnable runnable) {
         return new KafkaThread(name, runnable, true);

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTaskContext.java

@@ -35,7 +35,7 @@
 
 public class WorkerSinkTaskContext implements SinkTaskContext {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(WorkerSinkTaskContext.class);
     private final Map<TopicPartition, Long> offsets;
     private final Consumer<byte[], byte[]> consumer;
     private final WorkerSinkTask sinkTask;

metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java

@@ -163,7 +163,7 @@ public String toString() {
     }
 
     public static class SharedLogData {
-        private final Logger log = LoggerFactory.getLogger(SharedLogData.class);
+        private static final Logger log = LoggerFactory.getLogger(SharedLogData.class);
 
         /**
          * Maps node IDs to the matching log managers.

server-common/src/main/java/org/apache/kafka/server/util/TopicFilter.java

@@ -54,7 +54,7 @@ public Pattern getPattern() {
     }
 
     public static class IncludeList extends TopicFilter {
-        private final Logger log = LoggerFactory.getLogger(IncludeList.class);
+        private static final Logger log = LoggerFactory.getLogger(IncludeList.class);
         public IncludeList(String rawRegex) {
             super(rawRegex);
         }

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java

@@ -43,7 +43,7 @@
 
 public class KStreamSlidingWindowAggregate<KIn, VIn, VAgg> implements KStreamAggProcessorSupplier<KIn, VIn, Windowed<KIn>, VAgg> {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(KStreamSlidingWindowAggregate.class);
 
     private final String storeName;
     private final SlidingWindows windows;

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java

@@ -41,7 +41,7 @@
 
 public class KStreamWindowAggregate<KIn, VIn, VAgg, W extends Window> implements KStreamAggProcessorSupplier<KIn, VIn, Windowed<KIn>, VAgg> {
 
-    private final Logger log = LoggerFactory.getLogger(getClass());
+    private static final Logger log = LoggerFactory.getLogger(KStreamWindowAggregate.class);
 
     private final String storeName;
     private final Windows<W> windows;

streams/src/main/java/org/apache/kafka/streams/kstream/internals/graph/StreamSourceNode.java

@@ -30,7 +30,7 @@
 
 public class StreamSourceNode<K, V> extends SourceGraphNode<K, V> {
 
-    private final Logger log = LoggerFactory.getLogger(StreamSourceNode.class);
+    private static final Logger log = LoggerFactory.getLogger(StreamSourceNode.class);
 
     public StreamSourceNode(final String nodeName,
                             final Collection<String> topicNames,

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java

@@ -45,7 +45,7 @@
 
 public class ProcessorNode<KIn, VIn, KOut, VOut> {
 
-    private final Logger log = LoggerFactory.getLogger(ProcessorNode.class);
+    private static final Logger log = LoggerFactory.getLogger(ProcessorNode.class);
     private final List<ProcessorNode<KOut, VOut, ?, ?>> children;
     private final Map<String, ProcessorNode<KOut, VOut, ?, ?>> childByName;
 

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorTopology.java

@@ -31,7 +31,7 @@
 import java.util.Set;
 
 public class ProcessorTopology {
-    private final Logger log = LoggerFactory.getLogger(ProcessorTopology.class);
+    private static final Logger log = LoggerFactory.getLogger(ProcessorTopology.class);
 
     private final List<ProcessorNode<?, ?, ?, ?>> processorNodes;
     private final Map<String, SourceNode<?, ?>> sourceNodesByName;

streams/src/main/java/org/apache/kafka/streams/state/internals/WindowStoreBuilder.java

@@ -29,7 +29,7 @@
 import java.util.Objects;
 
 public class WindowStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, WindowStore<K, V>> {
-    private final Logger log = LoggerFactory.getLogger(WindowStoreBuilder.class);
+    private static final Logger log = LoggerFactory.getLogger(WindowStoreBuilder.class);
 
     private final WindowBytesStoreSupplier storeSupplier;
 

streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionOptimizingTest.java

@@ -72,7 +72,7 @@
 @SuppressWarnings("deprecation")
 public class RepartitionOptimizingTest {
 
-    private final Logger log = LoggerFactory.getLogger(RepartitionOptimizingTest.class);
+    private static final Logger log = LoggerFactory.getLogger(RepartitionOptimizingTest.class);
 
     private static final String INPUT_TOPIC = "input";
     private static final String COUNT_TOPIC = "outputTopic_0";

streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionWithMergeOptimizingTest.java

@@ -60,7 +60,7 @@
 
 public class RepartitionWithMergeOptimizingTest {
 
-    private final Logger log = LoggerFactory.getLogger(RepartitionWithMergeOptimizingTest.class);
+    private static final Logger log = LoggerFactory.getLogger(RepartitionWithMergeOptimizingTest.class);
 
     private static final String INPUT_A_TOPIC = "inputA";
     private static final String INPUT_B_TOPIC = "inputB";

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamsAssignmentScaleTest.java

@@ -82,7 +82,7 @@ public class StreamsAssignmentScaleTest {
     static final long MAX_ASSIGNMENT_DURATION = 120 * 1000L; // we should stay below `max.poll.interval.ms`
     static final String APPLICATION_ID = "streams-assignment-scale-test";
 
-    private final Logger log = LoggerFactory.getLogger(StreamsAssignmentScaleTest.class);
+    private static final Logger log = LoggerFactory.getLogger(StreamsAssignmentScaleTest.class);
 
     /* HighAvailabilityTaskAssignor tests */
 

streams/src/test/java/org/apache/kafka/streams/tests/StreamsUpgradeTest.java

@@ -154,7 +154,7 @@ public Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config) {
 
     public static class FutureStreamsPartitionAssignor extends StreamsPartitionAssignor {
         private static final Map<String, String> CLIENT_TAGS = mkMap(mkEntry("t1", "v1"), mkEntry("t2", "v2"));
-        private final Logger log = LoggerFactory.getLogger(FutureStreamsPartitionAssignor.class);
+        private static final Logger log = LoggerFactory.getLogger(FutureStreamsPartitionAssignor.class);
 
         private AtomicInteger usedSubscriptionMetadataVersionPeek;
         private AtomicLong nextScheduledRebalanceMs;

trogdor/src/main/java/org/apache/kafka/trogdor/workload/TimestampRecordProcessor.java

@@ -55,7 +55,7 @@ public class TimestampRecordProcessor implements RecordProcessor {
     private final ByteBuffer buffer;
     private final Histogram histogram;
 
-    private final Logger log = LoggerFactory.getLogger(TimestampRecordProcessor.class);
+    private static final Logger log = LoggerFactory.getLogger(TimestampRecordProcessor.class);
 
     static final float[] PERCENTILES = {0.5f, 0.95f, 0.99f};