diff --git a/spring-kafka-test/src/main/java/org/apache/kafka/test/TestCondition.java b/spring-kafka-test/src/main/java/org/apache/kafka/test/TestCondition.java new file mode 100644 index 00000000..f78c91bf --- /dev/null +++ b/spring-kafka-test/src/main/java/org/apache/kafka/test/TestCondition.java @@ -0,0 +1,26 @@ +/** + * 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.test; + +/** + * Interface to wrap actions that are required to wait until a condition is met + * for testing purposes. Note that this is not intended to do any assertions. + */ +public interface TestCondition { + + boolean conditionMet(); +} diff --git a/spring-kafka-test/src/main/java/org/apache/kafka/test/TestUtils.java b/spring-kafka-test/src/main/java/org/apache/kafka/test/TestUtils.java new file mode 100644 index 00000000..4904a5d3 --- /dev/null +++ b/spring-kafka-test/src/main/java/org/apache/kafka/test/TestUtils.java @@ -0,0 +1,262 @@ +/** + * 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.test; + +import static java.util.Arrays.asList; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Random; +import java.util.UUID; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.Cluster; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.record.CompressionType; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.record.Record; +import org.apache.kafka.common.record.Records; +import org.apache.kafka.common.utils.Utils; + +/** + * Helper functions for writing unit tests + */ +public class TestUtils { + + public static final File IO_TMP_DIR = new File(System.getProperty("java.io.tmpdir")); + + public static final String LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"; + public static final String DIGITS = "0123456789"; + public static final String LETTERS_AND_DIGITS = LETTERS + DIGITS; + + /* A consistent random number generator to make tests repeatable */ + public static final Random SEEDED_RANDOM = new Random(192348092834L); + public static final Random RANDOM = new Random(); + public static final long DEFAULT_MAX_WAIT_MS = 15000; + + public static Cluster singletonCluster(Map topicPartitionCounts) { + return clusterWith(1, topicPartitionCounts); + } + + public static Cluster singletonCluster(String topic, int partitions) { + return clusterWith(1, topic, partitions); + } + + public static Cluster clusterWith(int nodes, Map topicPartitionCounts) { + Node[] ns = new Node[nodes]; + for (int i = 0; i < nodes; i++) { + ns[i] = new Node(i, "localhost", 1969); + } + List parts = new ArrayList<>(); + for (Map.Entry topicPartition : topicPartitionCounts.entrySet()) { + String topic = topicPartition.getKey(); + int partitions = topicPartition.getValue(); + for (int i = 0; i < partitions; i++) { + parts.add(new PartitionInfo(topic, i, ns[i % ns.length], ns, ns)); + } + } + return new Cluster(asList(ns), parts, Collections.emptySet()); + } + + public static Cluster clusterWith(int nodes, String topic, int partitions) { + return clusterWith(nodes, Collections.singletonMap(topic, partitions)); + } + + /** + * Generate an array of random bytes + * + * @param size The size of the array + */ + public static byte[] randomBytes(int size) { + byte[] bytes = new byte[size]; + SEEDED_RANDOM.nextBytes(bytes); + return bytes; + } + + /** + * Generate a random string of letters and digits of the given length + * + * @param len The length of the string + * @return The random string + */ + public static String randomString(int len) { + StringBuilder b = new StringBuilder(); + for (int i = 0; i < len; i++) { + b.append(LETTERS_AND_DIGITS.charAt(SEEDED_RANDOM.nextInt(LETTERS_AND_DIGITS.length()))); + } + return b.toString(); + } + + /** + * Create an empty file in the default temporary-file directory, using `kafka` as the prefix and `tmp` as the + * suffix to generate its name. + */ + public static File tempFile() throws IOException { + File file = File.createTempFile("kafka", ".tmp"); + file.deleteOnExit(); + + return file; + } + + /** + * Create a temporary relative directory in the default temporary-file directory with the given prefix. + * + * @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix + */ + public static File tempDirectory(String prefix) { + return tempDirectory(null, prefix); + } + + /** + * Create a temporary relative directory in the default temporary-file directory with a + * prefix of "kafka-" + * @return the temporary directory just created. + */ + public static File tempDirectory() { + return tempDirectory(null); + } + + /** + * Create a temporary relative directory in the specified parent directory with the given prefix. + * + * @param parent The parent folder path name, if null using the default temporary-file directory + * @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix + */ + public static File tempDirectory(Path parent, String prefix) { + final File file; + prefix = prefix == null ? "kafka-" : prefix; + try { + file = parent == null ? + Files.createTempDirectory(prefix).toFile() : Files.createTempDirectory(parent, prefix).toFile(); + } catch (IOException ex) { + throw new RuntimeException("Failed to create a temp dir", ex); + } + file.deleteOnExit(); + + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + Utils.delete(file); + } + }); + + return file; + } + + /** + * Create a records buffer including the offset and message size at the start, which is required if the buffer is to + * be sent as part of `ProduceRequest`. This is the reason why we can't use + * `Record(long timestamp, byte[] key, byte[] value, CompressionType type, int valueOffset, int valueSize)` as this + * constructor does not include either of these fields. + */ + public static ByteBuffer partitionRecordsBuffer(long offset, CompressionType compressionType, Record... records) { + int bufferSize = 0; + for (Record record : records) { + bufferSize += Records.LOG_OVERHEAD + record.size(); + } + ByteBuffer buffer = ByteBuffer.allocate(bufferSize); + MemoryRecords memoryRecords = MemoryRecords.emptyRecords(buffer, compressionType); + for (Record record : records) { + memoryRecords.append(offset, record); + } + memoryRecords.close(); + return memoryRecords.buffer(); + } + + public static Properties producerConfig(final String bootstrapServers, + final Class keySerializer, + final Class valueSerializer, + final Properties additional) { + final Properties properties = new Properties(); + properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + properties.put(ProducerConfig.ACKS_CONFIG, "all"); + properties.put(ProducerConfig.RETRIES_CONFIG, 0); + properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer); + properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer); + properties.putAll(additional); + return properties; + } + + public static Properties producerConfig(final String bootstrapServers, Class keySerializer, Class valueSerializer) { + return producerConfig(bootstrapServers, keySerializer, valueSerializer, new Properties()); + } + + public static Properties consumerConfig(final String bootstrapServers, + final String groupId, + final Class keyDeserializer, + final Class valueDeserializer, + final Properties additional) { + + final Properties consumerConfig = new Properties(); + consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); + consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); + consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer); + consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer); + consumerConfig.putAll(additional); + return consumerConfig; + } + + /** + * returns consumer config with random UUID for the Group ID + */ + public static Properties consumerConfig(final String bootstrapServers, Class keyDeserializer, Class valueDeserializer) { + return consumerConfig(bootstrapServers, + UUID.randomUUID().toString(), + keyDeserializer, + valueDeserializer, + new Properties()); + } + + /** + * uses default value of 15 seconds for timeout + */ + public static void waitForCondition(TestCondition testCondition, String conditionDetails) throws InterruptedException { + waitForCondition(testCondition, DEFAULT_MAX_WAIT_MS, conditionDetails); + } + + /** + * Wait for condition to be met for at most {@code maxWaitMs} and throw assertion failure otherwise. + * This should be used instead of {@code Thread.sleep} whenever possible as it allows a longer timeout to be used + * without unnecessarily increasing test time (as the condition is checked frequently). The longer timeout is needed to + * avoid transient failures due to slow or overloaded machines. + */ + public static void waitForCondition(TestCondition testCondition, long maxWaitMs, String conditionDetails) throws InterruptedException { + long startTime = System.currentTimeMillis(); + + + while (!testCondition.conditionMet() && ((System.currentTimeMillis() - startTime) < maxWaitMs)) { + Thread.sleep(Math.min(maxWaitMs, 100L)); + } + + if (!testCondition.conditionMet()) { + conditionDetails = conditionDetails != null ? conditionDetails : ""; + throw new AssertionError("Condition not met within timeout " + maxWaitMs + ". " + conditionDetails); + } + } + +}