content: /*
   * 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 org.apache.kafka.clients.MetadataSnapshot;
  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.TopicPartition;
  import org.apache.kafka.common.feature.Features;
  import org.apache.kafka.common.feature.SupportedVersionRange;
  import org.apache.kafka.common.message.ApiMessageType;
  import org.apache.kafka.common.message.ApiVersionsResponseData;
  import org.apache.kafka.common.network.NetworkReceive;
  import org.apache.kafka.common.network.Send;
  import org.apache.kafka.common.protocol.ApiKeys;
  import org.apache.kafka.common.protocol.Errors;
  import org.apache.kafka.common.record.UnalignedRecords;
  import org.apache.kafka.common.requests.ApiVersionsResponse;
  import org.apache.kafka.common.requests.ByteBufferChannel;
  import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata;
  import org.apache.kafka.common.requests.RequestHeader;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.utils.Exit;
  import org.apache.kafka.common.utils.Utils;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.File;
  import java.io.IOException;
  import java.lang.reflect.Field;
  import java.nio.ByteBuffer;
  import java.nio.charset.StandardCharsets;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.util.ArrayList;
  import java.util.Base64;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Properties;
  import java.util.Random;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.CancellationException;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.function.Consumer;
  import java.util.function.Supplier;
  import java.util.regex.Matcher;
  import java.util.regex.Pattern;
  import java.util.stream.Collectors;
  
  import static java.util.Arrays.asList;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  import static org.junit.jupiter.api.Assertions.fail;
  
  /**
   * Helper functions for writing unit tests
   */
  public class TestUtils {
      private static final Logger log = LoggerFactory.getLogger(TestUtils.class);
  
      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_POLL_INTERVAL_MS = 100;
      public static final long DEFAULT_MAX_WAIT_MS = 15000;
  
      public static Cluster singletonCluster() {
          return clusterWith(1);
      }
  
      public static Cluster singletonCluster(final String topic, final int partitions) {
          return clusterWith(1, topic, partitions);
      }
  
      public static Cluster clusterWith(int nodes) {
          return clusterWith(nodes, new HashMap<>());
      }
  
      public static Cluster clusterWith(final int nodes, final Map<String, Integer> topicPartitionCounts) {
          final Node[] ns = new Node[nodes];
          for (int i = 0; i < nodes; i++)
              ns[i] = new Node(i, "localhost", 1969);
          final List<PartitionInfo> parts = new ArrayList<>();
          for (final Map.Entry<String, Integer> topicPartition : topicPartitionCounts.entrySet()) {
              final String topic = topicPartition.getKey();
              final 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("kafka-cluster", asList(ns), parts, Collections.emptySet(), Collections.emptySet());
      }
  
      public static Cluster clusterWith(final int nodes, final String topic, final int partitions) {
          return clusterWith(nodes, Collections.singletonMap(topic, partitions));
      }
  
      /**
       * Test utility function to get MetadataSnapshot with configured nodes and partitions.
       * @param nodes number of nodes in the cluster
       * @param topicPartitionCounts map of topic -> # of partitions
       * @return a MetadataSnapshot with number of nodes, partitions as per the input.
       */
  
      public static MetadataSnapshot metadataSnapshotWith(final int nodes, final Map<String, Integer> topicPartitionCounts) {
          final Node[] ns = new Node[nodes];
          Map<Integer, Node> nodesById = new HashMap<>();
          for (int i = 0; i < nodes; i++) {
              ns[i] = new Node(i, "localhost", 1969);
              nodesById.put(ns[i].id(), ns[i]);
          }
          final List<PartitionMetadata> partsMetadatas = new ArrayList<>();
          for (final Map.Entry<String, Integer> topicPartition : topicPartitionCounts.entrySet()) {
              final String topic = topicPartition.getKey();
              final int partitions = topicPartition.getValue();
              for (int i = 0; i < partitions; i++) {
                  TopicPartition tp = new TopicPartition(topic, partitions);
                  Node node = ns[i % ns.length];
                  partsMetadatas.add(new PartitionMetadata(Errors.NONE, tp, Optional.of(node.id()), Optional.empty(), null, null, null));
              }
          }
          return new MetadataSnapshot("kafka-cluster", nodesById, partsMetadatas, Collections.emptySet(), Collections.emptySet(), Collections.emptySet(), null, Collections.emptyMap());
      }
  
      /**
       * Asserts that there are no leaked threads with a specified name prefix and daemon status.
       * This method checks all threads in the JVM, filters them by the provided thread name prefix
       * and daemon status, and verifies that no matching threads are alive.
       * If any matching threads are found, the test will fail.
       *
       * @param threadName The prefix of the thread names to check. Only threads whose names
       *                   start with this prefix will be considered.
       * @param isDaemon   The daemon status to check. Only threads with the specified
       *                   daemon status (either true for daemon threads or false for non-daemon threads)
       *                   will be considered.
       *
       * @throws AssertionError If any thread with the specified name prefix and daemon status is found and is alive.
       */
      public static void assertNoLeakedThreadsWithNameAndDaemonStatus(String threadName, boolean isDaemon) {
          List<Thread> threads = Thread.getAllStackTraces().keySet().stream()
                  .filter(t -> t.isDaemon() == isDaemon && t.isAlive() && t.getName().startsWith(threadName))
                  .collect(Collectors.toList());
          int threadCount = threads.size();
          assertEquals(0, threadCount);
      }
  
      /**
       * Test utility function to get MetadataSnapshot of cluster with configured, and 0 partitions.
       * @param nodes number of nodes in the cluster.
       * @return a MetadataSnapshot of cluster with number of nodes in the input.
       */
      public static MetadataSnapshot metadataSnapshotWith(int nodes) {
          return metadataSnapshotWith(nodes, new HashMap<>());
      }
  
      /**
       * Generate an array of random bytes
       *
       * @param size The size of the array
       */
      public static byte[] randomBytes(final int size) {
          final 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(final int len) {
          final 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 the given prefix and suffix
       * to generate its name.
       * @throws IOException
       */
      public static File tempFile(final String prefix, final String suffix) throws IOException {
          final File file = Files.createTempFile(prefix, suffix).toFile();
          file.deleteOnExit();
          return file;
      }
  
      /**
       * 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 {
          return tempFile("kafka", ".tmp");
      }
  
      /**
       * Create a file with the given contents in the default temporary-file directory,
       * using `kafka` as the prefix and `tmp` as the suffix to generate its name.
       */
      public static File tempFile(final String contents) throws IOException {
          final File file = tempFile();
          Files.write(file.toPath(), contents.getBytes(StandardCharsets.UTF_8));
          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(final 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 directory under the given root directory.
       * The root directory is removed on JVM exit if it doesn't already exist
       * when this function is invoked.
       *
       * @param root path to create temporary directory under
       * @return the temporary directory created within {@code root}
       */
      public static File tempRelativeDir(String root) {
          File rootFile = new File(root);
          if (rootFile.mkdir()) {
              rootFile.deleteOnExit();
          }
          return tempDirectory(rootFile.toPath(), 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(final 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 (final IOException ex) {
              throw new RuntimeException("Failed to create a temp dir", ex);
          }
  
          Exit.addShutdownHook("delete-temp-file-shutdown-hook", () -> {
              try {
                  Utils.delete(file);
              } catch (IOException e) {
                  log.error("Error deleting {}", file.getAbsolutePath(), e);
              }
          });
  
          return file;
      }
  
      /**
       * Create a random log directory in the format <string>-<int> used for Kafka partition logs.
       * It is the responsibility of the caller to set up a shutdown hook for deletion of the directory.
       */
      public static File randomPartitionLogDir(File parentDir) {
          int attempts = 1000;
          while (attempts > 0) {
              File f = new File(parentDir, "kafka-" + RANDOM.nextInt(1000000));
              if (f.mkdir()) {
                  f.deleteOnExit();
                  return f;
              }
              attempts--;
          }
          throw new RuntimeException("Failed to create directory after 1000 attempts");
      }
  
      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.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, final Class<?> keySerializer, final Class<?> valueSerializer) {
          return producerConfig(bootstrapServers, keySerializer, valueSerializer, new Properties());
      }
  
      public static Properties requiredConsumerConfig() {
          final Properties consumerConfig = new Properties();
          consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
          consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          return consumerConfig;
      }
  
      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;
      }
  
      public static Properties consumerConfig(final String bootstrapServers,
                                              final String groupId,
                                              final Class<?> keyDeserializer,
                                              final Class<?> valueDeserializer) {
          return consumerConfig(bootstrapServers,
              groupId,
              keyDeserializer,
              valueDeserializer,
              new Properties());
      }
  
      /**
       * returns consumer config with random UUID for the Group ID
       */
      public static Properties consumerConfig(final String bootstrapServers, final Class<?> keyDeserializer, final Class<?> valueDeserializer) {
          return consumerConfig(bootstrapServers,
              UUID.randomUUID().toString(),
              keyDeserializer,
              valueDeserializer,
              new Properties());
      }
  
      /**
       * uses default value of 15 seconds for timeout
       */
      public static void waitForCondition(final TestCondition testCondition, final String conditionDetails) throws InterruptedException {
          waitForCondition(testCondition, DEFAULT_MAX_WAIT_MS, () -> conditionDetails);
      }
  
      /**
       * uses default value of 15 seconds for timeout
       */
      public static void waitForCondition(final TestCondition testCondition, final Supplier<String> conditionDetailsSupplier) throws InterruptedException {
          waitForCondition(testCondition, DEFAULT_MAX_WAIT_MS, conditionDetailsSupplier);
      }
  
      /**
       * 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(final TestCondition testCondition, final long maxWaitMs, String conditionDetails) throws InterruptedException {
          waitForCondition(testCondition, maxWaitMs, () -> 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(final TestCondition testCondition, final long maxWaitMs, Supplier<String> conditionDetailsSupplier) throws InterruptedException {
          waitForCondition(testCondition, maxWaitMs, DEFAULT_POLL_INTERVAL_MS, conditionDetailsSupplier);
      }
  
      /**
       * Wait for condition to be met for at most {@code maxWaitMs} with a polling interval of {@code pollIntervalMs}
       * 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(
          final TestCondition testCondition,
          final long maxWaitMs,
          final long pollIntervalMs,
          Supplier<String> conditionDetailsSupplier
      ) throws InterruptedException {
          retryOnExceptionWithTimeout(maxWaitMs, pollIntervalMs, () -> {
              String conditionDetailsSupplied = conditionDetailsSupplier != null ? conditionDetailsSupplier.get() : null;
              String conditionDetails = conditionDetailsSupplied != null ? conditionDetailsSupplied : "";
              assertTrue(testCondition.conditionMet(),
                  "Condition not met within timeout " + maxWaitMs + ". " + conditionDetails);
          });
      }
  
      /**
       * Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
       * {@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
       * last exception or assertion failure will be thrown thus providing context for the failure.
       *
       * @param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
       * @param runnable the code to attempt to execute successfully.
       * @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
       */
      public static void retryOnExceptionWithTimeout(final long timeoutMs,
                                                     final ValuelessCallable runnable) throws InterruptedException {
          retryOnExceptionWithTimeout(timeoutMs, DEFAULT_POLL_INTERVAL_MS, runnable);
      }
  
      /**
       * Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
       * {@link AssertionError}s, or for the default timeout to expire. If the timeout expires then the
       * last exception or assertion failure will be thrown thus providing context for the failure.
       *
       * @param runnable the code to attempt to execute successfully.
       * @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
       */
      public static void retryOnExceptionWithTimeout(final ValuelessCallable runnable) throws InterruptedException {
          retryOnExceptionWithTimeout(DEFAULT_MAX_WAIT_MS, DEFAULT_POLL_INTERVAL_MS, runnable);
      }
  
      /**
       * Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
       * {@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
       * last exception or assertion failure will be thrown thus providing context for the failure.
       *
       * @param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
       * @param pollIntervalMs the interval in milliseconds to wait between invoking {@code runnable}.
       * @param runnable the code to attempt to execute successfully.
       * @throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
       */
      public static void retryOnExceptionWithTimeout(final long timeoutMs,
                                                     final long pollIntervalMs,
                                                     final ValuelessCallable runnable) throws InterruptedException {
          final long expectedEnd = System.currentTimeMillis() + timeoutMs;
  
          while (true) {
              try {
                  runnable.call();
                  return;
              } catch (final NoRetryException e) {
                  throw e;
              } catch (final AssertionError t) {
                  if (expectedEnd <= System.currentTimeMillis()) {
                      throw t;
                  }
              } catch (final Exception e) {
                  if (expectedEnd <= System.currentTimeMillis()) {
                      throw new AssertionError(String.format("Assertion failed with an exception after %s ms", timeoutMs), e);
                  }
              }
              Thread.sleep(Math.min(pollIntervalMs, timeoutMs));
          }
      }
  
      /**
       * Checks if a cluster id is valid.
       * @param clusterId
       */
      public static void isValidClusterId(String clusterId) {
          assertNotNull(clusterId);
  
          // Base 64 encoded value is 22 characters
          assertEquals(22, clusterId.length());
  
          Pattern clusterIdPattern = Pattern.compile("[a-zA-Z0-9_\\-]+");
          Matcher matcher = clusterIdPattern.matcher(clusterId);
          assertTrue(matcher.matches());
  
          // Convert into normal variant and add padding at the end.
          String originalClusterId = String.format("%s==", clusterId.replace("_", "/").replace("-", "+"));
          byte[] decodedUuid = Base64.getDecoder().decode(originalClusterId);
  
          // We expect 16 bytes, same as the input UUID.
          assertEquals(16, decodedUuid.length);
  
          //Check if it can be converted back to a UUID.
          try {
              ByteBuffer uuidBuffer = ByteBuffer.wrap(decodedUuid);
              new UUID(uuidBuffer.getLong(), uuidBuffer.getLong()).toString();
          } catch (Exception e) {
              fail(clusterId + " cannot be converted back to UUID.");
          }
      }
  
      /**
       * Checks the two iterables for equality by first converting both to a list.
       */
      public static <T> void checkEquals(Iterable<T> it1, Iterable<T> it2) {
          assertEquals(toList(it1), toList(it2));
      }
  
      public static <T> void checkEquals(Iterator<T> it1, Iterator<T> it2) {
          assertEquals(Utils.toList(it1), Utils.toList(it2));
      }
  
      public static <T> void checkEquals(Set<T> c1, Set<T> c2, String firstDesc, String secondDesc) {
          if (!c1.equals(c2)) {
              Set<T> missing1 = new HashSet<>(c2);
              missing1.removeAll(c1);
              Set<T> missing2 = new HashSet<>(c1);
              missing2.removeAll(c2);
              fail(String.format("Sets not equal, missing %s=%s, missing %s=%s", firstDesc, missing1, secondDesc, missing2));
          }
      }
  
      public static <T> List<T> toList(Iterable<? extends T> iterable) {
          List<T> list = new ArrayList<>();
          for (T item : iterable)
              list.add(item);
          return list;
      }
  
      public static ByteBuffer toBuffer(Send send) {
          ByteBufferChannel channel = new ByteBufferChannel(send.size());
          try {
              assertEquals(send.size(), send.writeTo(channel));
              channel.close();
              return channel.buffer();
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
      }
  
      public static ByteBuffer toBuffer(UnalignedRecords records) {
          return toBuffer(records.toSend());
      }
  
      /**
       * Assert that a future raises an expected exception cause type.
       * This method will wait for the future to complete or timeout(15000 milliseconds).
       *
       * @param <T> Exception cause type parameter
       * @param exceptionCauseClass Class of the expected exception cause
       * @param future The future to await
       * @return The caught exception cause
       */
      public static <T extends Throwable> T assertFutureThrows(Class<T> exceptionCauseClass, Future<?> future) {
          try {
              future.get(DEFAULT_MAX_WAIT_MS, TimeUnit.MILLISECONDS);
              fail("Should throw expected exception " + exceptionCauseClass.getSimpleName() + " but nothing was thrown.");
          } catch (InterruptedException | ExecutionException | CancellationException e) {
              Throwable cause = e instanceof ExecutionException ? e.getCause() : e;
              // Enable strict type checking.
              // This ensures we're testing for the exact exception type, not its subclasses.
              assertEquals(
                  exceptionCauseClass, 
                  cause.getClass(), 
                  "Expected " + exceptionCauseClass.getSimpleName() + ", but got " + cause.getClass().getSimpleName()
              );
              return exceptionCauseClass.cast(cause);
          } catch (TimeoutException e) {
              fail("Future is not completed within " + DEFAULT_MAX_WAIT_MS + " milliseconds.");
          } catch (Exception e) {
              fail("Expected " + exceptionCauseClass.getSimpleName() + ", but got " + e.getClass().getSimpleName());
          }
          return null;
      }
  
      public static <T extends Throwable> void assertFutureThrows(
          Class<T> expectedCauseClassApiException,
          Future<?> future,
          String expectedMessage
      ) {
          T receivedException = assertFutureThrows(expectedCauseClassApiException, future);
          assertEquals(expectedMessage, receivedException.getMessage());
      }
  
      public static ApiKeys apiKeyFrom(NetworkReceive networkReceive) {
          return RequestHeader.parse(networkReceive.payload().duplicate()).apiKey();
      }
  
      public static <T> void assertOptional(Optional<T> optional, Consumer<T> assertion) {
          if (optional.isPresent()) {
              assertion.accept(optional.get());
          } else {
              fail("Missing value from Optional");
          }
      }
  
      @SuppressWarnings("unchecked")
      public static <T> T fieldValue(Object o, Class<?> clazz, String fieldName)  {
          try {
              Field field = clazz.getDeclaredField(fieldName);
              field.setAccessible(true);
              return (T) field.get(o);
          } catch (Exception e) {
              throw new RuntimeException(e);
          }
      }
  
      public static void setFieldValue(Object obj, String fieldName, Object value) throws Exception {
          Field field = obj.getClass().getDeclaredField(fieldName);
          field.setAccessible(true);
          field.set(obj, value);
      }
  
      /**
       * Returns true if both iterators have same elements in the same order.
       *
       * @param iterator1 first iterator.
       * @param iterator2 second iterator.
       * @param <T>       type of element in the iterators.
       */
      public static <T> boolean sameElementsWithOrder(Iterator<T> iterator1,
                                                      Iterator<T> iterator2) {
          while (iterator1.hasNext()) {
              if (!iterator2.hasNext()) {
                  return false;
              }
  
              if (!Objects.equals(iterator1.next(), iterator2.next())) {
                  return false;
              }
          }
  
          return !iterator2.hasNext();
      }
  
      /**
       * Returns true if both the iterators have same set of elements irrespective of order and duplicates.
       *
       * @param iterator1 first iterator.
       * @param iterator2 second iterator.
       * @param <T>       type of element in the iterators.
       */
      public static <T> boolean sameElementsWithoutOrder(Iterator<T> iterator1,
                                                         Iterator<T> iterator2) {
          // Check both the iterators have the same set of elements irrespective of order and duplicates.
          Set<T> allSegmentsSet = new HashSet<>();
          iterator1.forEachRemaining(allSegmentsSet::add);
          Set<T> expectedSegmentsSet = new HashSet<>();
          iterator2.forEachRemaining(expectedSegmentsSet::add);
  
          return allSegmentsSet.equals(expectedSegmentsSet);
      }
  
      public static ApiVersionsResponse defaultApiVersionsResponse(
              ApiMessageType.ListenerType listenerType
      ) {
          return defaultApiVersionsResponse(0, listenerType);
      }
  
      public static ApiVersionsResponse defaultApiVersionsResponse(
              int throttleTimeMs,
              ApiMessageType.ListenerType listenerType
      ) {
          return createApiVersionsResponse(
                  throttleTimeMs,
                  ApiVersionsResponse.filterApis(listenerType, true, true),
                  Features.emptySupportedFeatures(),
                  false
          );
      }
  
      public static ApiVersionsResponse defaultApiVersionsResponse(
              int throttleTimeMs,
              ApiMessageType.ListenerType listenerType,
              boolean enableUnstableLastVersion
      ) {
          return createApiVersionsResponse(
                  throttleTimeMs,
                  ApiVersionsResponse.filterApis(listenerType, enableUnstableLastVersion, true),
                  Features.emptySupportedFeatures(),
                  false
          );
      }
  
      public static ApiVersionsResponse createApiVersionsResponse(
              int throttleTimeMs,
              ApiVersionsResponseData.ApiVersionCollection apiVersions
      ) {
          return createApiVersionsResponse(throttleTimeMs, apiVersions, Features.emptySupportedFeatures(), false);
      }
  
      public static ApiVersionsResponse createApiVersionsResponse(
              int throttleTimeMs,
              ApiVersionsResponseData.ApiVersionCollection apiVersions,
              Features<SupportedVersionRange> latestSupportedFeatures,
              boolean zkMigrationEnabled
      ) {
          return new ApiVersionsResponse.Builder().
              setThrottleTimeMs(throttleTimeMs).
              setApiVersions(apiVersions).
              setSupportedFeatures(latestSupportedFeatures).
              setFinalizedFeatures(Collections.emptyMap()).
              setFinalizedFeaturesEpoch(ApiVersionsResponse.UNKNOWN_FINALIZED_FEATURES_EPOCH).
              setZkMigrationEnabled(zkMigrationEnabled).
              build();
      }
  }
  