MINOR: remove unused helper methods from utils classes

clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallback.java

@@ -24,8 +24,6 @@
 import java.util.Map;
 import java.util.Objects;
 
-import static org.apache.kafka.common.utils.CollectionUtils.subtractMap;
-
 /**
  * A {@code Callback} for use by the {@code SaslServer} implementation when it
  * needs to validate the SASL extensions for the OAUTHBEARER mechanism
@@ -82,13 +80,6 @@ public Map<String, String> invalidExtensions() {
         return Collections.unmodifiableMap(invalidExtensions);
     }
 
-    /**
-     * @return An immutable {@link Map} consisting of the extensions that have neither been validated nor invalidated
-     */
-    public Map<String, String> ignoredExtensions() {
-        return Collections.unmodifiableMap(subtractMap(subtractMap(inputExtensions.map(), invalidExtensions), validatedExtensions));
-    }
-
     /**
      * Validates a specific extension in the original {@code inputExtensions} map
      * @param extensionName - the name of the extension which was validated

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

@@ -23,39 +23,11 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Collection;
-import java.util.stream.Collectors;
 
 public final class CollectionUtils {
 
     private CollectionUtils() {}
 
-    /**
-     * Given two maps (A, B), returns all the key-value pairs in A whose keys are not contained in B
-     */
-    public static <K, V> Map<K, V> subtractMap(Map<? extends K, ? extends V> minuend, Map<? extends K, ? extends V> subtrahend) {
-        return minuend.entrySet().stream()
-                .filter(entry -> !subtrahend.containsKey(entry.getKey()))
-                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-    }
-
-    /**
-     * group data by topic
-     *
-     * @param data Data to be partitioned
-     * @param <T> Partition data type
-     * @return partitioned data
-     */
-    public static <T> Map<String, Map<Integer, T>> groupPartitionDataByTopic(Map<TopicPartition, ? extends T> data) {
-        Map<String, Map<Integer, T>> dataByTopic = new HashMap<>();
-        for (Map.Entry<TopicPartition, ? extends T> entry : data.entrySet()) {
-            String topic = entry.getKey().topic();
-            int partition = entry.getKey().partition();
-            Map<Integer, T> topicData = dataByTopic.computeIfAbsent(topic, t -> new HashMap<>());
-            topicData.put(partition, entry.getValue());
-        }
-        return dataByTopic;
-    }
-
     /**
      * Group a list of partitions by the topic name.
      *

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

@@ -72,8 +72,6 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collector;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
@@ -781,21 +779,7 @@ public static <K, V> Map<K, V> mkMap(final Map.Entry<K, V>... entries) {
      * @param properties A map of properties to add
      * @return The properties object
      */
-    public static Properties mkProperties(final Map<String, String> properties) {
-        final Properties result = new Properties();
-        for (final Map.Entry<String, String> entry : properties.entrySet()) {
-            result.setProperty(entry.getKey(), entry.getValue());
-        }
-        return result;
-    }
-
-    /**
-     * Creates a {@link Properties} from a map
-     *
-     * @param properties A map of properties to add
-     * @return The properties object
-     */
-    public static Properties mkObjectProperties(final Map<String, Object> properties) {
+    public static Properties mkProperties(final Map<String, Object> properties) {
         final Properties result = new Properties();
         for (final Map.Entry<String, Object> entry : properties.entrySet()) {
             result.put(entry.getKey(), entry.getValue());
@@ -861,15 +845,6 @@ public FileVisitResult postVisitDirectory(Path path, IOException exc) throws IOE
         });
     }
 
-    /**
-     * Returns an empty collection if this list is null
-     * @param other
-     * @return
-     */
-    public static <T> List<T> safe(List<T> other) {
-        return other == null ? Collections.emptyList() : other;
-    }
-
    /**
     * Get the ClassLoader which loaded Kafka.
     */
@@ -934,18 +909,6 @@ public static void closeAll(Closeable... closeables) throws IOException {
             throw exception;
     }
 
-    /**
-     * An {@link AutoCloseable} interface without a throws clause in the signature
-     *
-     * This is used with lambda expressions in try-with-resources clauses
-     * to avoid casting un-checked exceptions to checked exceptions unnecessarily.
-     */
-    @FunctionalInterface
-    public interface UncheckedCloseable extends AutoCloseable {
-        @Override
-        void close();
-    }
-
     /**
      * Closes {@code closeable} and if an exception is thrown, it is logged at the WARN level.
      */
@@ -1151,15 +1114,6 @@ public static <T> List<T> toList(Iterator<T> iterator) {
         return res;
     }
 
-    public static <T> List<T> concatListsUnmodifiable(List<T> left, List<T> right) {
-        return concatLists(left, right, Collections::unmodifiableList);
-    }
-
-    public static <T> List<T> concatLists(List<T> left, List<T> right, Function<List<T>, List<T>> finisher) {
-        return Stream.concat(left.stream(), right.stream())
-                .collect(Collectors.collectingAndThen(Collectors.toList(), finisher));
-    }
-
     public static int to32BitField(final Set<Byte> bytes) {
         int value = 0;
         for (final byte b : bytes)
@@ -1185,18 +1139,6 @@ public static Set<Byte> from32BitField(final int intValue) {
         return result;
     }
 
-    public static <K1, V1, K2, V2> Map<K2, V2> transformMap(
-            Map<? extends K1, ? extends V1> map,
-            Function<K1, K2> keyMapper,
-            Function<V1, V2> valueMapper) {
-        return map.entrySet().stream().collect(
-            Collectors.toMap(
-                entry -> keyMapper.apply(entry.getKey()),
-                entry -> valueMapper.apply(entry.getValue())
-            )
-        );
-    }
-
     /**
      * A Collector that offers two kinds of convenience:
      * 1. You can specify the concrete type of the returned Map

clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallbackTest.java

@@ -76,7 +76,7 @@ public void testUnvalidatedExtensionsAreIgnored() {
 
         assertFalse(callback.validatedExtensions().containsKey("nothing"));
         assertFalse(callback.invalidExtensions().containsKey("nothing"));
-        assertEquals("nothing", callback.ignoredExtensions().get("nothing"));
+        assertEquals("nothing", callback.inputExtensions().map().get("nothing"));
     }
 
     @Test

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

@@ -33,7 +33,6 @@
 import org.apache.kafka.common.metrics.stats.Value;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.common.utils.Utils.UncheckedCloseable;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.errors.RetriableException;
@@ -196,12 +195,13 @@ public void execute() {
         initializeAndStart();
         // Make sure any uncommitted data has been committed and the task has
         // a chance to clean up its state
-        try (UncheckedCloseable suppressible = this::closePartitions) {
-            while (!isStopping())
-                iteration();
+        try {
+            while (!isStopping()) iteration();
         } catch (WakeupException e) {
-            log.trace("Consumer woken up during initial offset commit attempt, " 
-                + "but succeeded during a later attempt");
+            log.trace("Consumer woken up during initial offset commit attempt, "
+                    + "but succeeded during a later attempt");
+        } finally {
+            closePartitions();
         }
     }
 

streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java

@@ -49,7 +49,7 @@
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
+import static org.apache.kafka.common.utils.Utils.mkProperties;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.purgeLocalStreamsState;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
 import static org.apache.kafka.test.TestUtils.waitForCondition;
@@ -88,7 +88,7 @@ public void setup() {
         builder  = new StreamsBuilder();
         builder.stream(inputTopic);
 
-        properties  = mkObjectProperties(
+        properties = mkProperties(
             mkMap(
                 mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                 mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),

streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java

@@ -62,7 +62,6 @@
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
 import static org.apache.kafka.common.utils.Utils.mkProperties;
 import static org.apache.kafka.common.utils.Utils.mkSet;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
@@ -275,7 +274,7 @@ private static void assertFalseNoRetry(final boolean assertion, final String mes
 
     private static Properties streamsProperties(final String appId,
                                                 final AssignmentListener configuredAssignmentListener) {
-        return mkObjectProperties(
+        return mkProperties(
             mkMap(
                 mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                 mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),

streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java

@@ -56,7 +56,7 @@
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
+import static org.apache.kafka.common.utils.Utils.mkProperties;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
 import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
@@ -98,7 +98,7 @@ public void setup() {
         final KStream<String, String> stream = builder.stream(inputTopic);
         stream.process(() -> new ShutdownProcessor(processorValueCollector), Named.as("process"));
 
-        properties  = mkObjectProperties(
+        properties = mkProperties(
             mkMap(
                 mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                 mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),

streams/src/test/java/org/apache/kafka/streams/integration/TaskAssignorIntegrationTest.java

@@ -45,7 +45,7 @@
 
 import static org.apache.kafka.common.utils.Utils.mkEntry;
 import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.common.utils.Utils.mkObjectProperties;
+import static org.apache.kafka.common.utils.Utils.mkProperties;
 import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.instanceOf;
@@ -89,7 +89,7 @@ public void shouldProperlyConfigureTheAssignor() throws NoSuchFieldException, Il
         final AssignmentListener configuredAssignmentListener =
             stable -> compilerDefeatingReference.incrementAndGet();
 
-        final Properties properties = mkObjectProperties(
+        final Properties properties = mkProperties(
             mkMap(
                 mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()),
                 mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, appId),