diff --git a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallback.java b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallback.java index eab208baa6f2b..1207b2f62d51f 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallback.java +++ b/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 invalidExtensions() { return Collections.unmodifiableMap(invalidExtensions); } - /** - * @return An immutable {@link Map} consisting of the extensions that have neither been validated nor invalidated - */ - public Map 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 diff --git a/clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java b/clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java index f3a7dbd52bf08..88e3d4afe50ca 100644 --- a/clients/src/main/java/org/apache/kafka/common/utils/CollectionUtils.java +++ b/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 Map subtractMap(Map minuend, Map 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 Partition data type - * @return partitioned data - */ - public static Map> groupPartitionDataByTopic(Map data) { - Map> dataByTopic = new HashMap<>(); - for (Map.Entry entry : data.entrySet()) { - String topic = entry.getKey().topic(); - int partition = entry.getKey().partition(); - Map topicData = dataByTopic.computeIfAbsent(topic, t -> new HashMap<>()); - topicData.put(partition, entry.getValue()); - } - return dataByTopic; - } - /** * Group a list of partitions by the topic name. * diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java index 7e46d51c32be9..6000f14fd0aea 100755 --- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java +++ b/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 Map mkMap(final Map.Entry... entries) { * @param properties A map of properties to add * @return The properties object */ - public static Properties mkProperties(final Map properties) { - final Properties result = new Properties(); - for (final Map.Entry 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 properties) { + public static Properties mkProperties(final Map properties) { final Properties result = new Properties(); for (final Map.Entry 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 List safe(List 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 List toList(Iterator iterator) { return res; } - public static List concatListsUnmodifiable(List left, List right) { - return concatLists(left, right, Collections::unmodifiableList); - } - - public static List concatLists(List left, List right, Function, List> finisher) { - return Stream.concat(left.stream(), right.stream()) - .collect(Collectors.collectingAndThen(Collectors.toList(), finisher)); - } - public static int to32BitField(final Set bytes) { int value = 0; for (final byte b : bytes) @@ -1185,18 +1139,6 @@ public static Set from32BitField(final int intValue) { return result; } - public static Map transformMap( - Map map, - Function keyMapper, - Function 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 diff --git a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallbackTest.java b/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallbackTest.java index 82ff4fa82da1b..a470d68dc94fb 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/oauthbearer/OAuthBearerExtensionsValidatorCallbackTest.java +++ b/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 diff --git a/clients/src/test/java/org/apache/kafka/common/utils/CollectionUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/CollectionUtilsTest.java deleted file mode 100644 index 7f8419e28a2b3..0000000000000 --- a/clients/src/test/java/org/apache/kafka/common/utils/CollectionUtilsTest.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.kafka.common.utils; - -import org.junit.jupiter.api.Test; - -import java.util.HashMap; -import java.util.Map; - -import static org.apache.kafka.common.utils.CollectionUtils.subtractMap; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.junit.jupiter.api.Assertions.assertNotSame; - -public class CollectionUtilsTest { - - @Test - public void testSubtractMapRemovesSecondMapsKeys() { - Map mainMap = new HashMap<>(); - mainMap.put("one", "1"); - mainMap.put("two", "2"); - mainMap.put("three", "3"); - Map secondaryMap = new HashMap<>(); - secondaryMap.put("one", "4"); - secondaryMap.put("two", "5"); - - Map newMap = subtractMap(mainMap, secondaryMap); - - assertEquals(3, mainMap.size()); // original map should not be modified - assertEquals(1, newMap.size()); - assertTrue(newMap.containsKey("three")); - assertEquals("3", newMap.get("three")); - } - - @Test - public void testSubtractMapDoesntRemoveAnythingWhenEmptyMap() { - Map mainMap = new HashMap<>(); - mainMap.put("one", "1"); - mainMap.put("two", "2"); - mainMap.put("three", "3"); - Map secondaryMap = new HashMap<>(); - - Map newMap = subtractMap(mainMap, secondaryMap); - - assertEquals(3, newMap.size()); - assertEquals("1", newMap.get("one")); - assertEquals("2", newMap.get("two")); - assertEquals("3", newMap.get("three")); - assertNotSame(newMap, mainMap); - } -} 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 index 1229fec7204b1..5dd3d1886df0c 100644 --- 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 @@ -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(); } } diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java index 764af9c524cff..bfc7bc4afe52f 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/AdjustStreamThreadCountTest.java +++ b/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), diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java index 530a8544844da..d6e9b81766e3d 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/HighAvailabilityTaskAssignorIntegrationTest.java +++ b/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), diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java index eed626feee21f..3698610acd377 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java +++ b/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 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), diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/TaskAssignorIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/TaskAssignorIntegrationTest.java index 6046d9245990d..2ff4f299271fc 100644 --- a/streams/src/test/java/org/apache/kafka/streams/integration/TaskAssignorIntegrationTest.java +++ b/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),