MINOR: Cleanups in connect:runtime (#19452)

Various cleanups in Connect:
- use enhanced switch
- remove dead code
- convert classes to records

Reviewers: Chia-Ping Tsai <[email protected]>, TengYao Chi
 <[email protected]>, Ken Huang <[email protected]>

Author: mimaison

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

@@ -73,7 +73,6 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.PrintStream;
-import java.io.UnsupportedEncodingException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -1082,12 +1081,8 @@ protected final boolean maybeAddConfigErrors(
 
     private String trace(Throwable t) {
         ByteArrayOutputStream output = new ByteArrayOutputStream();
-        try {
-            t.printStackTrace(new PrintStream(output, false, StandardCharsets.UTF_8.name()));
-            return output.toString(StandardCharsets.UTF_8.name());
-        } catch (UnsupportedEncodingException e) {
-            return null;
-        }
+        t.printStackTrace(new PrintStream(output, false, StandardCharsets.UTF_8));
+        return output.toString(StandardCharsets.UTF_8);
     }
 
     /*

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

@@ -54,7 +54,7 @@ private void validate(String connectorName, VersionRange range) throws ConnectEx
         }
     }
 
-    private Connector lookup(String connectorName, VersionRange range) throws Exception {
+    private Connector lookup(String connectorName, VersionRange range) {
         String version = range == null ? LATEST_VERSION : range.toString();
         if (connectors.containsKey(connectorName) && connectors.get(connectorName).containsKey(version)) {
             return connectors.get(connectorName).get(version);
@@ -73,7 +73,7 @@ private Connector lookup(String connectorName, VersionRange range) throws Except
         }
     }
 
-    public Connector getConnector(String connectorName, VersionRange range) throws Exception {
+    public Connector getConnector(String connectorName, VersionRange range) {
         validate(connectorName, range);
         return lookup(connectorName, range);
     }

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

@@ -372,7 +372,7 @@ public <R extends ConnectRecord<R>> List<TransformationStage<R>> transformationS
             try {
                 final String typeConfig = prefix + "type";
                 final String versionConfig = prefix + WorkerConfig.PLUGIN_VERSION_SUFFIX;
-                @SuppressWarnings("unchecked") final Transformation<R> transformation = getTransformationOrPredicate(plugins, typeConfig, versionConfig);
+                final Transformation<R> transformation = getTransformationOrPredicate(plugins, typeConfig, versionConfig);
                 Map<String, Object> configs = originalsWithPrefix(prefix);
                 Object predicateAlias = configs.remove(TransformationStage.PREDICATE_CONFIG);
                 Object negate = configs.remove(TransformationStage.NEGATE_CONFIG);
@@ -382,7 +382,6 @@ public <R extends ConnectRecord<R>> List<TransformationStage<R>> transformationS
                     String predicatePrefix = PREDICATES_PREFIX + predicateAlias + ".";
                     final String predicateTypeConfig = predicatePrefix + "type";
                     final String predicateVersionConfig = predicatePrefix + WorkerConfig.PLUGIN_VERSION_SUFFIX;
-                    @SuppressWarnings("unchecked")
                     Predicate<R> predicate = getTransformationOrPredicate(plugins, predicateTypeConfig, predicateVersionConfig);
                     predicate.configure(originalsWithPrefix(predicatePrefix));
                     Plugin<Predicate<R>> predicatePlugin = metrics.wrap(predicate, connectorTaskId, (String) predicateAlias);
@@ -781,22 +780,7 @@ public boolean visible(String name, Map<String, Object> parsedConfig) {
         }
     }
 
-    private static class ConverterDefaults {
-        private final String type;
-        private final String version;
-
-        public ConverterDefaults(String type, String version) {
-            this.type = type;
-            this.version = version;
-        }
-
-        public String type() {
-            return type;
-        }
-
-        public String version() {
-            return version;
-        }
+    private record ConverterDefaults(String type, String version) {
     }
 
     public static class PluginVersionValidator implements ConfigDef.Validator {

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

@@ -428,24 +428,22 @@ private TransactionBoundaryManager buildTransactionManager(
             SourceConnectorConfig sourceConfig,
             WorkerTransactionContext transactionContext) {
         TransactionBoundary boundary = sourceConfig.transactionBoundary();
-        switch (boundary) {
-            case POLL:
-                return new TransactionBoundaryManager() {
-                    @Override
-                    protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
-                        return true;
-                    }
-
-                    @Override
-                    protected boolean shouldCommitFinalTransaction() {
-                        return true;
-                    }
-                };
+        return switch (boundary) {
+            case POLL -> new TransactionBoundaryManager() {
+                @Override
+                protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
+                    return true;
+                }
 
-            case INTERVAL:
+                @Override
+                protected boolean shouldCommitFinalTransaction() {
+                    return true;
+                }
+            };
+            case INTERVAL -> {
                 long transactionBoundaryInterval = Optional.ofNullable(sourceConfig.transactionBoundaryInterval())
                         .orElse(workerConfig.offsetCommitInterval());
-                return new TransactionBoundaryManager() {
+                yield new TransactionBoundaryManager() {
                     private final long commitInterval = transactionBoundaryInterval;
                     private long lastCommit;
 
@@ -465,14 +463,14 @@ protected boolean shouldCommitTransactionForBatch(long currentTimeMs) {
                     }
 
                     @Override
-                    protected  boolean shouldCommitFinalTransaction() {
+                    protected boolean shouldCommitFinalTransaction() {
                         return true;
                     }
                 };
-
-            case CONNECTOR:
+            }
+            case CONNECTOR -> {
                 Objects.requireNonNull(transactionContext, "Transaction context must be provided when using connector-defined transaction boundaries");
-                return new TransactionBoundaryManager() {
+                yield new TransactionBoundaryManager() {
                     @Override
                     protected boolean shouldCommitFinalTransaction() {
                         return shouldCommitTransactionForBatch(time.milliseconds());
@@ -513,9 +511,8 @@ private void maybeAbortTransaction() {
                         transactionOpen = false;
                     }
                 };
-            default:
-                throw new IllegalArgumentException("Unrecognized transaction boundary: " + boundary);
-        }
+            }
+        };
     }
 
     TransactionMetricsGroup transactionMetricsGroup() {

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

@@ -37,7 +37,6 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 
 /**
  * <p>
@@ -392,35 +391,6 @@ enum ConfigReloadAction {
         RESTART
     }
 
-    class Created<T> {
-        private final boolean created;
-        private final T result;
-
-        public Created(boolean created, T result) {
-            this.created = created;
-            this.result = result;
-        }
-
-        public boolean created() {
-            return created;
-        }
-
-        public T result() {
-            return result;
-        }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-            Created<?> created1 = (Created<?>) o;
-            return Objects.equals(created, created1.created) &&
-                    Objects.equals(result, created1.result);
-        }
-
-        @Override
-        public int hashCode() {
-            return Objects.hash(created, result);
-        }
+    record Created<T>(boolean created, T result) {
     }
 }

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

@@ -22,9 +22,7 @@
 import org.apache.kafka.connect.util.ConnectorTaskId;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Objects;
-import java.util.stream.Collectors;
 
 /**
  * An immutable restart plan per connector.
@@ -45,13 +43,10 @@ public RestartPlan(RestartRequest request, ConnectorStateInfo restartStateInfo)
         this.request = Objects.requireNonNull(request, "RestartRequest name may not be null");
         this.stateInfo = Objects.requireNonNull(restartStateInfo, "ConnectorStateInfo name may not be null");
         // Collect the task IDs to stop and restart (may be none)
-        this.idsToRestart = Collections.unmodifiableList(
-                stateInfo.tasks()
-                        .stream()
-                        .filter(this::isRestarting)
-                        .map(taskState -> new ConnectorTaskId(request.connectorName(), taskState.id()))
-                        .collect(Collectors.toList())
-        );
+        this.idsToRestart = stateInfo.tasks()
+                .stream()
+                .filter(this::isRestarting)
+                .map(taskState -> new ConnectorTaskId(request.connectorName(), taskState.id())).toList();
     }
 
     /**

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

@@ -17,61 +17,25 @@
 package org.apache.kafka.connect.runtime;
 
 import org.apache.kafka.connect.connector.Connector;
-import org.apache.kafka.connect.connector.Task;
 
 import java.util.Objects;
 
 /**
  * A request to restart a connector and/or task instances.
  * <p>
  * The natural order is based first upon the connector name and then requested restart behaviors.
- * If two requests have the same connector name, then the requests are ordered based on the 
+ * If two requests have the same connector name, then the requests are ordered based on the
  * probable number of tasks/connector this request is going to restart.
+ * @param connectorName the name of the connector; may not be null
+ * @param onlyFailed    true if only failed instances should be restarted
+ * @param includeTasks  true if tasks should be restarted, or false if only the connector should be restarted
  */
-public class RestartRequest implements Comparable<RestartRequest> {
+public record RestartRequest(String connectorName,
+                             boolean onlyFailed,
+                             boolean includeTasks) implements Comparable<RestartRequest> {
 
-    private final String connectorName;
-    private final boolean onlyFailed;
-    private final boolean includeTasks;
-
-    /**
-     * Create a new request to restart a connector and optionally its tasks.
-     *
-     * @param connectorName the name of the connector; may not be null
-     * @param onlyFailed    true if only failed instances should be restarted
-     * @param includeTasks  true if tasks should be restarted, or false if only the connector should be restarted
-     */
-    public RestartRequest(String connectorName, boolean onlyFailed, boolean includeTasks) {
-        this.connectorName = Objects.requireNonNull(connectorName, "Connector name may not be null");
-        this.onlyFailed = onlyFailed;
-        this.includeTasks = includeTasks;
-    }
-
-    /**
-     * Get the name of the connector.
-     *
-     * @return the connector name; never null
-     */
-    public String connectorName() {
-        return connectorName;
-    }
-
-    /**
-     * Determine whether only failed instances be restarted.
-     *
-     * @return true if only failed instances should be restarted, or false if all applicable instances should be restarted
-     */
-    public boolean onlyFailed() {
-        return onlyFailed;
-    }
-
-    /**
-     * Determine whether {@link Task} instances should also be restarted in addition to the {@link Connector} instance.
-     *
-     * @return true if the connector and task instances should be restarted, or false if just the connector should be restarted
-     */
-    public boolean includeTasks() {
-        return includeTasks;
+    public RestartRequest {
+        Objects.requireNonNull(connectorName, "Connector name may not be null");
     }
 
     /**
@@ -108,6 +72,7 @@ public int compareTo(RestartRequest o) {
         int result = connectorName.compareTo(o.connectorName);
         return result == 0 ? impactRank() - o.impactRank() : result;
     }
+
     //calculates an internal rank for the restart request based on the probable number of tasks/connector this request is going to restart
     private int impactRank() {
         if (onlyFailed && !includeTasks) { //restarts only failed connector so least impactful
@@ -121,23 +86,6 @@ private int impactRank() {
         return 3;
     }
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        RestartRequest that = (RestartRequest) o;
-        return onlyFailed == that.onlyFailed && includeTasks == that.includeTasks && Objects.equals(connectorName, that.connectorName);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(connectorName, onlyFailed, includeTasks);
-    }
-
     @Override
     public String toString() {
         return "restart request for {" + "connectorName='" + connectorName + "', onlyFailed=" + onlyFailed + ", includeTasks=" + includeTasks + '}';

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

@@ -22,53 +22,12 @@
 
 /**
  * A session key, which can be used to validate internal REST requests between workers.
+ * @param key               the actual cryptographic key to use for request validation; may not be null
+ * @param creationTimestamp the time at which the key was generated
  */
-public class SessionKey {
+public record SessionKey(SecretKey key, long creationTimestamp) {
 
-    private final SecretKey key;
-    private final long creationTimestamp;
-
-    /**
-     * Create a new session key with the given key value and creation timestamp
-     * @param key the actual cryptographic key to use for request validation; may not be null
-     * @param creationTimestamp the time at which the key was generated
-     */
-    public SessionKey(SecretKey key, long creationTimestamp) {
-        this.key = Objects.requireNonNull(key, "Key may not be null");
-        this.creationTimestamp = creationTimestamp;
-    }
-
-    /**
-     * Get the cryptographic key to use for request validation.
-     *
-     * @return the cryptographic key; may not be null
-     */
-    public SecretKey key() {
-        return key;
-    }
-
-    /**
-     * Get the time at which the key was generated.
-     *
-     * @return the time at which the key was generated
-     */
-    public long creationTimestamp() {
-        return creationTimestamp;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o)
-            return true;
-        if (o == null || getClass() != o.getClass())
-            return false;
-        SessionKey that = (SessionKey) o;
-        return creationTimestamp == that.creationTimestamp
-            && key.equals(that.key);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(key, creationTimestamp);
+    public SessionKey {
+        Objects.requireNonNull(key, "Key may not be null");
     }
 }