Fix spotbugs and checkstyle errors

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -588,7 +588,7 @@ static KafkaAdminClient createInternal(AdminClientConfig config,
             metrics = new Metrics(new MetricConfig(), new LinkedList<>(), time);
             LogContext logContext = createLogContext(clientId);
             return new KafkaAdminClient(config, clientId, time, metadataManager, metrics,
-                client, null, logContext, null);
+                client, null, logContext, Optional.empty());
         } catch (Throwable exc) {
             closeQuietly(metrics, "Metrics");
             throw new KafkaException("Failed to create new KafkaAdminClient", exc);

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -1547,7 +1547,7 @@ public void registerMetric(KafkaMetric metric) {
      */
     @Override
     public void unregisterMetric(KafkaMetric metric) {
-         delegate.unregisterMetric(metric);
+        delegate.unregisterMetric(metric);
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -1326,10 +1326,10 @@ public void registerMetricForSubscription(KafkaMetric metric) {
      */
     @Override
     public void unregisterMetricFromSubscription(KafkaMetric metric) {
-         if (clientTelemetryReporter.isPresent()) {
-             ClientTelemetryReporter reporter = clientTelemetryReporter.get();
-             reporter.metricRemoval(metric);
-         }
+        if (clientTelemetryReporter.isPresent()) {
+            ClientTelemetryReporter reporter = clientTelemetryReporter.get();
+            reporter.metricRemoval(metric);
+        }
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java

@@ -90,13 +90,13 @@ public class MockProducer<K, V> implements Producer<K, V> {
     /**
      * Create a mock producer
      *
-     * @param cluster The cluster holding metadata for this producer
-     * @param autoComplete If true automatically complete all requests successfully and execute the callback. Otherwise
-     *        the user must call {@link #completeNext()} or {@link #errorNext(RuntimeException)} after
-     *        {@link #send(ProducerRecord) send()} to complete the call and unblock the {@link
-     *        java.util.concurrent.Future Future&lt;RecordMetadata&gt;} that is returned.
-     * @param partitioner The partition strategy
-     * @param keySerializer The serializer for key that implements {@link Serializer}.
+     * @param cluster         The cluster holding metadata for this producer
+     * @param autoComplete    If true automatically complete all requests successfully and execute the callback. Otherwise
+     *                        the user must call {@link #completeNext()} or {@link #errorNext(RuntimeException)} after
+     *                        {@link #send(ProducerRecord) send()} to complete the call and unblock the {@link
+     *                        java.util.concurrent.Future Future&lt;RecordMetadata&gt;} that is returned.
+     * @param partitioner     The partition strategy
+     * @param keySerializer   The serializer for key that implements {@link Serializer}.
      * @param valueSerializer The serializer for value that implements {@link Serializer}.
      */
     public MockProducer(final Cluster cluster,
@@ -120,7 +120,7 @@ public MockProducer(final Cluster cluster,
 
     /**
      * Create a new mock producer with invented metadata the given autoComplete setting and key\value serializers.
-     *
+     * <p>
      * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(Cluster.empty(), autoComplete, new DefaultPartitioner(), keySerializer, valueSerializer)}
      */
     @SuppressWarnings("deprecation")
@@ -132,7 +132,7 @@ public MockProducer(final boolean autoComplete,
 
     /**
      * Create a new mock producer with invented metadata the given autoComplete setting and key\value serializers.
-     *
+     * <p>
      * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(cluster, autoComplete, new DefaultPartitioner(), keySerializer, valueSerializer)}
      */
     @SuppressWarnings("deprecation")
@@ -145,7 +145,7 @@ public MockProducer(final Cluster cluster,
 
     /**
      * Create a new mock producer with invented metadata the given autoComplete setting, partitioner and key\value serializers.
-     *
+     * <p>
      * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(Cluster.empty(), autoComplete, partitioner, keySerializer, valueSerializer)}
      */
     public MockProducer(final boolean autoComplete,
@@ -157,7 +157,7 @@ public MockProducer(final boolean autoComplete,
 
     /**
      * Create a new mock producer with invented metadata.
-     *
+     * <p>
      * Equivalent to {@link #MockProducer(Cluster, boolean, Partitioner, Serializer, Serializer) new MockProducer(Cluster.empty(), false, null, null, null)}
      */
     public MockProducer() {
@@ -226,7 +226,7 @@ public void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offs
             return;
         }
         Map<TopicPartition, OffsetAndMetadata> uncommittedOffsets =
-            this.uncommittedConsumerGroupOffsets.computeIfAbsent(groupMetadata.groupId(), k -> new HashMap<>());
+                this.uncommittedConsumerGroupOffsets.computeIfAbsent(groupMetadata.groupId(), k -> new HashMap<>());
         uncommittedOffsets.putAll(offsets);
         this.sentOffsets = true;
     }
@@ -336,7 +336,7 @@ public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record, Cal
             keySerializer.serialize(record.topic(), record.key());
             valueSerializer.serialize(record.topic(), record.value());
         }
-            
+
         TopicPartition topicPartition = new TopicPartition(record.topic(), partition);
         ProduceRequestResult result = new ProduceRequestResult(topicPartition);
         FutureRecordMetadata future = new FutureRecordMetadata(result, 0, RecordBatch.NO_TIMESTAMP,
@@ -504,7 +504,6 @@ public synchronized List<ProducerRecord<K, V>> uncommittedRecords() {
     }
 
     /**
-     *
      * Get the list of committed consumer group offsets since the last call to {@link #clear()}
      */
     public synchronized List<Map<String, Map<TopicPartition, OffsetAndMetadata>>> consumerGroupOffsetsHistory() {
@@ -611,11 +610,11 @@ public void complete(RuntimeException e) {
 
     @Override
     public void registerMetricForSubscription(KafkaMetric metric) {
-          addedMetrics.add(metric);
+        addedMetrics.add(metric);
     }
 
     @Override
     public void unregisterMetricFromSubscription(KafkaMetric metric) {
-          addedMetrics.remove(metric);
+        addedMetrics.remove(metric);
     }
 }

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

@@ -33,7 +33,6 @@
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.internals.KafkaFutureImpl;
-import org.apache.kafka.common.metrics.KafkaMetric;
 import org.apache.kafka.common.metrics.Metrics;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -66,14 +65,11 @@
 import org.slf4j.Logger;
 
 import java.time.Duration;
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;