MINOR: Various cleanups in connect (apache#15734)

Reviewers: Chia-Ping Tsai <[email protected]>
dongnuo123 · Apr 17, 2024 · 7b1c777 · 7b1c777
1 parent 363f4d2
commit 7b1c777
Show file tree

Hide file tree

Showing 54 changed files with 97 additions and 235 deletions.
diff --git a/...main/java/org/apache/kafka/connect/rest/basic/auth/extension/PropertyFileLoginModule.java b/...main/java/org/apache/kafka/connect/rest/basic/auth/extension/PropertyFileLoginModule.java
@@ -48,14 +48,13 @@
  */
 public class PropertyFileLoginModule implements LoginModule {
     private static final Logger log = LoggerFactory.getLogger(PropertyFileLoginModule.class);
-    private static final Map<String, Properties> CREDENTIAL_PROPERTIES_MAP = new ConcurrentHashMap<>();
     private static final String FILE_OPTIONS = "file";
+    private static final Map<String, Properties> CREDENTIAL_PROPERTIES = new ConcurrentHashMap<>();
+
     private CallbackHandler callbackHandler;
     private String fileName;
     private boolean authenticated;
 
-    private static final Map<String, Properties> CREDENTIAL_PROPERTIES = new ConcurrentHashMap<>();
-
     @Override
     public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState, Map<String, ?> options) {
         this.callbackHandler = callbackHandler;

diff --git a/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSinkTaskTest.java b/connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSinkTaskTest.java
@@ -56,7 +56,7 @@ public void setup() {
     @Test
     public void testPutFlush() {
         HashMap<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
-        final String newLine = System.getProperty("line.separator"); 
+        final String newLine = System.lineSeparator();
 
         // We do not call task.start() since it would override the output stream
 

diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConnector.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointConnector.java
@@ -203,10 +203,10 @@ Set<String> findConsumerGroups()
                     .collect(Collectors.toSet());
             // Only perform checkpoints for groups that have offsets for at least one topic that's accepted
             // by the topic filter.
-            if (consumedTopics.size() > 0) {
-                checkpointGroups.add(group);
-            } else {
+            if (consumedTopics.isEmpty()) {
                 irrelevantGroups.add(group);
+            } else {
+                checkpointGroups.add(group);
             }
         }
 

diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java
@@ -117,7 +117,6 @@ public class MirrorMaker {
     private final Time time;
     private final MirrorMakerConfig config;
     private final Set<String> clusters;
-    private final Set<SourceAndTarget> herderPairs;
     private final MirrorRestServer internalServer;
     private final RestClient restClient;
 
@@ -149,13 +148,13 @@ public MirrorMaker(MirrorMakerConfig config, List<String> clusters, Time time) {
             this.clusters = config.clusters();
         }
         log.info("Targeting clusters {}", this.clusters);
-        this.herderPairs = config.clusterPairs().stream()
+        Set<SourceAndTarget> herderPairs = config.clusterPairs().stream()
             .filter(x -> this.clusters.contains(x.target()))
             .collect(Collectors.toSet());
         if (herderPairs.isEmpty()) {
             throw new IllegalArgumentException("No source->target replication flows.");
         }
-        this.herderPairs.forEach(this::addHerder);
+        herderPairs.forEach(this::addHerder);
         shutdownHook = new ShutdownHook();
     }
 

diff --git a/.../src/main/java/org/apache/kafka/connect/mirror/rest/resources/InternalMirrorResource.java b/.../src/main/java/org/apache/kafka/connect/mirror/rest/resources/InternalMirrorResource.java
@@ -21,8 +21,6 @@
 import org.apache.kafka.connect.runtime.rest.RestClient;
 import org.apache.kafka.connect.runtime.rest.RestRequestTimeout;
 import org.apache.kafka.connect.runtime.rest.resources.InternalClusterResource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import javax.inject.Inject;
 import javax.ws.rs.NotFoundException;
@@ -37,8 +35,6 @@ public class InternalMirrorResource extends InternalClusterResource {
     @Context
     private UriInfo uriInfo;
 
-    private static final Logger log = LoggerFactory.getLogger(InternalMirrorResource.class);
-
     private final Map<SourceAndTarget, Herder> herders;
 
     @Inject

diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConnectorTest.java
@@ -91,13 +91,7 @@
 
 public class MirrorSourceConnectorTest {
     private ConfigPropertyFilter getConfigPropertyFilter() {
-        return new ConfigPropertyFilter() {
-            @Override
-            public boolean shouldReplicateConfigProperty(String prop) {
-                return true;
-            }
-
-        };
+        return prop -> true;
     }
 
     @Test

diff --git a/.../connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java b/.../connect/mirror/integration/MirrorConnectorsWithCustomForwardingAdminIntegrationTest.java
@@ -331,17 +331,13 @@ public void testSyncTopicACLsUseProvidedForwardingAdmin() throws Exception {
     }
 
     void waitForTopicToPersistInFakeLocalMetadataStore(String topicName) throws InterruptedException {
-        waitForCondition(() -> {
-                return FakeLocalMetadataStore.containsTopic(topicName);
-            }, FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
+        waitForCondition(() -> FakeLocalMetadataStore.containsTopic(topicName), FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
             "Topic: " + topicName + " didn't get created in the FakeLocalMetadataStore"
         );
     }
 
     void waitForTopicConfigPersistInFakeLocalMetaDataStore(String topicName, String configName, String expectedConfigValue) throws InterruptedException {
-        waitForCondition(() -> {
-            return FakeLocalMetadataStore.topicConfig(topicName).getOrDefault(configName, "").equals(expectedConfigValue);
-            }, FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
+        waitForCondition(() -> FakeLocalMetadataStore.topicConfig(topicName).getOrDefault(configName, "").equals(expectedConfigValue), FAKE_LOCAL_METADATA_STORE_SYNC_DURATION_MS,
             "Topic: " + topicName + "'s configs don't have " + configName + ":" + expectedConfigValue
         );
     }

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java b/connect/runtime/src/main/java/org/apache/kafka/connect/cli/ConnectDistributed.java
@@ -36,8 +36,6 @@
 import org.apache.kafka.connect.storage.StatusBackingStore;
 import org.apache.kafka.connect.util.ConnectUtils;
 import org.apache.kafka.connect.util.SharedTopicAdmin;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
 import java.util.HashMap;
@@ -55,7 +53,6 @@
  * </p>
  */
 public class ConnectDistributed extends AbstractConnectCli<DistributedConfig> {
-    private static final Logger log = LoggerFactory.getLogger(ConnectDistributed.class);
 
     public ConnectDistributed(String... args) {
         super(args);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java
@@ -668,7 +668,7 @@ private static ConfigInfos validateClientOverrides(String connName,
 
                 ConfigValue configValue = new ConfigValue(prefix + validatedConfigValue.name(), validatedConfigValue.value(),
                                                           validatedConfigValue.recommendedValues(), validatedConfigValue.errorMessages());
-                if (configValue.errorMessages().size() > 0) {
+                if (!configValue.errorMessages().isEmpty()) {
                     errorCount++;
                 }
                 ConfigValueInfo configValueInfo = convertConfigValue(configValue, configKey != null ? configKey.type : null);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java
@@ -391,7 +391,7 @@ boolean sendRecords() {
         int processed = 0;
         recordBatch(toSend.size());
         final SourceRecordWriteCounter counter =
-                toSend.size() > 0 ? new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup) : null;
+                toSend.isEmpty() ? null : new SourceRecordWriteCounter(toSend.size(), sourceTaskMetricsGroup);
         for (final SourceRecord preTransformRecord : toSend) {
             ProcessingContext<SourceRecord> context = new ProcessingContext<>(preTransformRecord);
             final SourceRecord record = transformationChain.apply(context, preTransformRecord);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java
@@ -346,7 +346,7 @@ public static ConfigDef enrich(Plugins plugins, ConfigDef baseConfigDef, Map<Str
         ConfigDef newDef = new ConfigDef(baseConfigDef);
         new EnrichablePlugin<Transformation<?>>("Transformation", TRANSFORMS_CONFIG, TRANSFORMS_GROUP, (Class) Transformation.class,
                 props, requireFullConfig) {
-            @SuppressWarnings("rawtypes")
+
             @Override
             protected Set<PluginDesc<Transformation<?>>> plugins() {
                 return plugins.transformations();

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
@@ -203,7 +203,7 @@ void initialize() {
         }
     }
 
-    private boolean doStart() throws Throwable {
+    private boolean doStart() {
         try {
             switch (state) {
                 case STARTED:
@@ -235,12 +235,12 @@ private synchronized void onFailure(Throwable t) {
         this.state = State.FAILED;
     }
 
-    private void resume() throws Throwable {
+    private void resume() {
         if (doStart())
             statusListener.onResume(connName);
     }
 
-    private void start() throws Throwable {
+    private void start() {
         if (doStart())
             statusListener.onStartup(connName);
     }
@@ -392,7 +392,7 @@ void doTransitionTo(TargetState targetState, Callback<TargetState> stateChangeCa
         }
     }
 
-    private void doTransitionTo(TargetState targetState) throws Throwable {
+    private void doTransitionTo(TargetState targetState) {
         log.debug("{} Transition connector to {}", this, targetState);
         if (targetState == TargetState.PAUSED) {
             suspend(true);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
@@ -794,8 +794,6 @@ private void onPartitionsRemoved(Collection<TopicPartition> partitions, boolean
     }
 
     static class SinkTaskMetricsGroup {
-        private final ConnectorTaskId id;
-        private final ConnectMetrics metrics;
         private final MetricGroup metricGroup;
         private final Sensor sinkRecordRead;
         private final Sensor sinkRecordSend;
@@ -805,13 +803,10 @@ static class SinkTaskMetricsGroup {
         private final Sensor offsetCompletionSkip;
         private final Sensor putBatchTime;
         private final Sensor sinkRecordActiveCount;
-        private long activeRecords;
         private Map<TopicPartition, OffsetAndMetadata> consumedOffsets = new HashMap<>();
         private Map<TopicPartition, OffsetAndMetadata> committedOffsets = new HashMap<>();
 
         public SinkTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) {
-            this.metrics = connectMetrics;
-            this.id = id;
 
             ConnectMetricsRegistry registry = connectMetrics.registry();
             metricGroup = connectMetrics
@@ -855,7 +850,7 @@ public SinkTaskMetricsGroup(ConnectorTaskId id, ConnectMetrics connectMetrics) {
         void computeSinkRecordLag() {
             Map<TopicPartition, OffsetAndMetadata> consumed = this.consumedOffsets;
             Map<TopicPartition, OffsetAndMetadata> committed = this.committedOffsets;
-            activeRecords = 0L;
+            long activeRecords = 0L;
             for (Map.Entry<TopicPartition, OffsetAndMetadata> committedOffsetEntry : committed.entrySet()) {
                 final TopicPartition partition = committedOffsetEntry.getKey();
                 final OffsetAndMetadata consumedOffsetMeta = consumed.get(partition);

diff --git a/...me/src/main/java/org/apache/kafka/connect/runtime/distributed/RequestTargetException.java b/...me/src/main/java/org/apache/kafka/connect/runtime/distributed/RequestTargetException.java
@@ -30,16 +30,6 @@ public RequestTargetException(String s, String forwardUrl) {
         this.forwardUrl = forwardUrl;
     }
 
-    public RequestTargetException(String s, Throwable throwable, String forwardUrl) {
-        super(s, throwable);
-        this.forwardUrl = forwardUrl;
-    }
-
-    public RequestTargetException(Throwable throwable, String forwardUrl) {
-        super(throwable);
-        this.forwardUrl = forwardUrl;
-    }
-
     public String forwardUrl() {
         return forwardUrl;
     }

diff --git a/...ntime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java b/...ntime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java
@@ -46,7 +46,7 @@ public class DelegatingClassLoader extends URLClassLoader {
     private final ConcurrentMap<String, String> aliases;
 
     // Although this classloader does not load classes directly but rather delegates loading to a
-    // PluginClassLoader or its parent through its base class, because of the use of inheritance in
+    // PluginClassLoader or its parent through its base class, because of the use of inheritance
     // in the latter case, this classloader needs to also be declared as parallel capable to use
     // fine-grain locking when loading classes.
     static {

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
@@ -398,7 +398,7 @@ public URI advertisedUrl() {
         String advertisedHostname = config.advertisedHostName();
         if (advertisedHostname != null && !advertisedHostname.isEmpty())
             builder.host(advertisedHostname);
-        else if (serverConnector != null && serverConnector.getHost() != null && serverConnector.getHost().length() > 0)
+        else if (serverConnector != null && serverConnector.getHost() != null && !serverConnector.getHost().isEmpty())
             builder.host(serverConnector.getHost());
 
         Integer advertisedPort = config.advertisedPort();

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaConfigBackingStore.java
@@ -365,7 +365,7 @@ public KafkaConfigBackingStore(Converter converter, DistributedConfig config, Wo
 
         this.usesFencableWriter = config.transactionalLeaderEnabled();
         this.topic = config.getString(DistributedConfig.CONFIG_TOPIC_CONFIG);
-        if (this.topic == null || this.topic.trim().length() == 0)
+        if (this.topic == null || this.topic.trim().isEmpty())
             throw new ConfigException("Must specify topic for connector configuration.");
 
         configLog = setupAndCreateKafkaBasedLog(this.topic, config);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java
@@ -173,7 +173,7 @@ public KafkaOffsetBackingStore(Supplier<TopicAdmin> topicAdmin, Supplier<String>
     @Override
     public void configure(final WorkerConfig config) {
         String topic = config.getString(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG);
-        if (topic == null || topic.trim().length() == 0)
+        if (topic == null || topic.trim().isEmpty())
             throw new ConfigException("Offset storage topic must be specified");
 
         this.exactlyOnce = config.exactlyOnceSourceEnabled();
@@ -249,7 +249,7 @@ public void start() {
                         + "support for source connectors, or upgrade to a newer Kafka broker version.";
             } else {
                 message = "When " + ConsumerConfig.ISOLATION_LEVEL_CONFIG + "is set to "
-                        + IsolationLevel.READ_COMMITTED.toString()
+                        + IsolationLevel.READ_COMMITTED
                         + ", a Kafka broker version that allows admin clients to read consumer offsets is required. "
                         + "Please either reconfigure the worker or connector, or upgrade to a newer Kafka broker version.";
             }

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java
@@ -170,7 +170,7 @@ public KafkaStatusBackingStore(Time time, Converter converter, Supplier<TopicAdm
     @Override
     public void configure(final WorkerConfig config) {
         this.statusTopic = config.getString(DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG);
-        if (this.statusTopic == null || this.statusTopic.trim().length() == 0)
+        if (this.statusTopic == null || this.statusTopic.trim().isEmpty())
             throw new ConfigException("Must specify topic for connector status.");
 
         sendRetryExecutor = Executors.newSingleThreadExecutor(
@@ -298,7 +298,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
                 if (exception == null) return;
                 // TODO: retry more gracefully and not forever
                 if (exception instanceof RetriableException) {
-                    sendRetryExecutor.submit((Runnable) () -> kafkaLog.send(key, value, this));
+                    sendRetryExecutor.submit(() -> kafkaLog.send(key, value, this));
                 } else {
                     log.error("Failed to write status update", exception);
                 }
@@ -332,7 +332,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
                             return;
                     }
 
-                    sendRetryExecutor.submit((Runnable) () -> kafkaLog.send(key, value, this));
+                    sendRetryExecutor.submit(() -> kafkaLog.send(key, value, this));
                 } else {
                     log.error("Failed to write status update", exception);
                 }

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/MemoryOffsetBackingStore.java
@@ -19,8 +19,6 @@
 import org.apache.kafka.connect.runtime.WorkerConfig;
 import org.apache.kafka.connect.util.Callback;
 import org.apache.kafka.common.utils.ThreadUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
 import java.util.Collection;
@@ -38,7 +36,6 @@
  * background thread.
  */
 public abstract class MemoryOffsetBackingStore implements OffsetBackingStore {
-    private static final Logger log = LoggerFactory.getLogger(MemoryOffsetBackingStore.class);
 
     protected Map<ByteBuffer, ByteBuffer> data = new HashMap<>();
     protected ExecutorService executor;

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/tools/PredicateDoc.java b/connect/runtime/src/main/java/org/apache/kafka/connect/tools/PredicateDoc.java
@@ -40,22 +40,18 @@ private <P extends Predicate<?>> DocInfo(Class<P> predicateClass, String overvie
         }
     }
 
-    private static final List<DocInfo> PREDICATES;
-    static {
-        List<DocInfo> collect = new Plugins(Collections.emptyMap()).predicates().stream()
-            .map(p -> {
-                try {
-                    String overviewDoc = (String) p.pluginClass().getDeclaredField("OVERVIEW_DOC").get(null);
-                    ConfigDef configDef = (ConfigDef) p.pluginClass().getDeclaredField("CONFIG_DEF").get(null);
-                    return new DocInfo(p.pluginClass(), overviewDoc, configDef);
-                } catch (ReflectiveOperationException e) {
-                    throw new RuntimeException("Predicate class " + p.pluginClass().getName() + " lacks either a `public static final String OVERVIEW_DOC` or `public static final ConfigDef CONFIG_DEF`");
-                }
-            })
-            .sorted(Comparator.comparing(docInfo -> docInfo.predicateName))
-            .collect(Collectors.toList());
-        PREDICATES = collect;
-    }
+    private static final List<DocInfo> PREDICATES = new Plugins(Collections.emptyMap()).predicates().stream()
+        .map(p -> {
+            try {
+                String overviewDoc = (String) p.pluginClass().getDeclaredField("OVERVIEW_DOC").get(null);
+                ConfigDef configDef = (ConfigDef) p.pluginClass().getDeclaredField("CONFIG_DEF").get(null);
+                return new DocInfo(p.pluginClass(), overviewDoc, configDef);
+            } catch (ReflectiveOperationException e) {
+                throw new RuntimeException("Predicate class " + p.pluginClass().getName() + " lacks either a `public static final String OVERVIEW_DOC` or `public static final ConfigDef CONFIG_DEF`");
+            }
+        })
+        .sorted(Comparator.comparing(docInfo -> docInfo.predicateName))
+        .collect(Collectors.toList());
 
     private static void printPredicateHtml(PrintStream out, DocInfo docInfo) {
         out.println("<div id=\"" + docInfo.predicateName + "\">");

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/KafkaBasedLog.java
@@ -251,7 +251,7 @@ public void start() {
             throw new ConnectException(
                     "Must provide a TopicAdmin to KafkaBasedLog when consumer is configured with "
                             + ConsumerConfig.ISOLATION_LEVEL_CONFIG + " set to "
-                            + IsolationLevel.READ_COMMITTED.toString()
+                            + IsolationLevel.READ_COMMITTED
             );
         }
         initializer.accept(admin);

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/util/LoggingContext.java b/connect/runtime/src/main/java/org/apache/kafka/connect/util/LoggingContext.java
@@ -106,18 +106,6 @@ public static LoggingContext forConnector(String connectorName) {
         return context;
     }
 
-    /**
-     * Modify the current {@link MDC} logging context to set the {@link #CONNECTOR_CONTEXT connector context} to include the
-     * supplied connector name and the {@link Scope#VALIDATE} scope.
-     *
-     * @param connectorName the connector name
-     */
-    public static LoggingContext forValidation(String connectorName) {
-        LoggingContext context = new LoggingContext();
-        MDC.put(CONNECTOR_CONTEXT, prefixFor(connectorName, Scope.VALIDATE, null));
-        return context;
-    }
-
     /**
      * Modify the current {@link MDC} logging context to set the {@link #CONNECTOR_CONTEXT connector context} to include the
      * connector name and task number using the supplied {@link ConnectorTaskId}, and to set the scope to {@link Scope#TASK}.