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.streams.tests;
  
  import org.apache.kafka.clients.admin.Admin;
  import org.apache.kafka.clients.admin.ConsumerGroupDescription;
  import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
  import org.apache.kafka.clients.admin.StreamsGroupDescription;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.clients.consumer.OffsetAndMetadata;
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.ProducerConfig;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.common.IsolationLevel;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.errors.TimeoutException;
  import org.apache.kafka.common.serialization.ByteArrayDeserializer;
  import org.apache.kafka.common.serialization.IntegerDeserializer;
  import org.apache.kafka.common.serialization.IntegerSerializer;
  import org.apache.kafka.common.serialization.LongDeserializer;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import org.apache.kafka.common.serialization.StringSerializer;
  import org.apache.kafka.common.utils.Exit;
  import org.apache.kafka.common.utils.Utils;
  
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Properties;
  import java.util.Random;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.TimeUnit;
  
  public class EosTestDriver extends SmokeTestUtil {
  
      private static final int MAX_NUMBER_OF_KEYS = 20000;
      private static final long MAX_IDLE_TIME_MS = 600000L;
  
      private static volatile boolean isRunning = true;
      private static final CountDownLatch TERMINATED = new CountDownLatch(1);
  
      private static int numRecordsProduced = 0;
  
      private static synchronized void updateNumRecordsProduces(final int delta) {
          numRecordsProduced += delta;
      }
  
      static void generate(final String kafka) {
          Exit.addShutdownHook("streams-eos-test-driver-shutdown-hook", () -> {
              System.out.println("Terminating");
              isRunning = false;
  
              try {
                  if (TERMINATED.await(5L, TimeUnit.MINUTES)) {
                      System.out.println("Terminated");
                  } else {
                      System.out.println("Terminated with timeout");
                  }
              } catch (final InterruptedException swallow) {
                  swallow.printStackTrace(System.err);
                  System.out.println("Terminated with error");
              }
              System.err.flush();
              System.out.flush();
          });
  
          final Properties producerProps = new Properties();
          producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, "EosTest");
          producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
          producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
          producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
          producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
  
          final Map<Integer, List<Long>> offsets = new HashMap<>();
  
          try {
              try (final KafkaProducer<String, Integer> producer = new KafkaProducer<>(producerProps)) {
                  final Random rand = new Random(System.currentTimeMillis());
  
                  while (isRunning) {
                      final String key = "" + rand.nextInt(MAX_NUMBER_OF_KEYS);
                      final int value = rand.nextInt(10000);
  
                      final ProducerRecord<String, Integer> record = new ProducerRecord<>("data", key, value);
  
                      producer.send(record, (metadata, exception) -> {
                          if (exception != null) {
                              exception.printStackTrace(System.err);
                              System.err.flush();
                              if (exception instanceof TimeoutException) {
                                  try {
                                      // message == org.apache.kafka.common.errors.TimeoutException: Expiring 4 record(s) for data-0: 30004 ms has passed since last attempt plus backoff time
                                      final int expired = Integer.parseInt(exception.getMessage().split(" ")[2]);
                                      updateNumRecordsProduces(-expired);
                                  } catch (final Exception ignore) {
                                  }
                              }
                          } else {
                              offsets.getOrDefault(metadata.partition(), new LinkedList<>()).add(metadata.offset());
                          }
                      });
  
                      updateNumRecordsProduces(1);
                      if (numRecordsProduced % 1000 == 0) {
                          System.out.println(numRecordsProduced + " records produced");
                          System.out.flush();
                      }
                      Utils.sleep(rand.nextInt(10));
                  }
              }
              System.out.println("Producer closed: " + numRecordsProduced + " records produced");
              System.out.flush();
  
              // verify offsets
              for (final Map.Entry<Integer, List<Long>> offsetsOfPartition : offsets.entrySet()) {
                  offsetsOfPartition.getValue().sort(Long::compareTo);
                  for (int i = 0; i < offsetsOfPartition.getValue().size() - 1; ++i) {
                      if (offsetsOfPartition.getValue().get(i) != i) {
                          System.err.println("Offset for partition " + offsetsOfPartition.getKey() + " is not " + i + " as expected but " + offsetsOfPartition.getValue().get(i));
                          System.err.flush();
                      }
                  }
                  System.out.println("Max offset of partition " + offsetsOfPartition.getKey() + " is " + offsetsOfPartition.getValue().get(offsetsOfPartition.getValue().size() - 1));
              }
  
              final Properties props = new Properties();
              props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
              props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
              props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
              props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
              props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString());
  
              try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
                  final List<TopicPartition> partitions = getAllPartitions(consumer, "data");
                  System.out.println("Partitions: " + partitions);
                  System.out.flush();
                  consumer.assign(partitions);
                  consumer.seekToEnd(partitions);
  
                  for (final TopicPartition tp : partitions) {
                      System.out.println("End-offset for " + tp + " is " + consumer.position(tp));
                      System.out.flush();
                  }
              }
              System.out.flush();
          } finally {
              TERMINATED.countDown();
          }
      }
  
      public static void verify(final String kafka, final boolean withRepartitioning, final String groupProtocol) {
          final Properties props = new Properties();
          props.put(ConsumerConfig.CLIENT_ID_CONFIG, "verifier");
          props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
          props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
          props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
          props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.toString());
  
          try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
              verifyAllTransactionFinished(consumer, kafka, withRepartitioning);
          } catch (final Exception e) {
              e.printStackTrace(System.err);
              System.out.println("FAILED");
              return;
          }
  
          final Map<TopicPartition, Long> committedOffsets;
          try (final Admin adminClient = Admin.create(props)) {
              ensureStreamsApplicationDown(adminClient, groupProtocol);
  
              committedOffsets = getCommittedOffsets(adminClient, withRepartitioning);
          }
  
          final String[] allInputTopics;
          final String[] allOutputTopics;
          if (withRepartitioning) {
              allInputTopics = new String[] {"data", "repartition"};
              allOutputTopics = new String[] {"echo", "min", "sum", "repartition", "max", "cnt"};
          } else {
              allInputTopics = new String[] {"data"};
              allOutputTopics = new String[] {"echo", "min", "sum"};
          }
  
          final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> inputRecordsPerTopicPerPartition;
          try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
              final List<TopicPartition> partitions = getAllPartitions(consumer, allInputTopics);
              consumer.assign(partitions);
              consumer.seekToBeginning(partitions);
  
              inputRecordsPerTopicPerPartition = getRecords(consumer, committedOffsets, withRepartitioning, true);
          } catch (final Exception e) {
              e.printStackTrace(System.err);
              System.out.println("FAILED");
              return;
          }
  
          final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> outputRecordsPerTopicPerPartition;
          try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
              final List<TopicPartition> partitions = getAllPartitions(consumer, allOutputTopics);
              consumer.assign(partitions);
              consumer.seekToBeginning(partitions);
  
              outputRecordsPerTopicPerPartition = getRecords(consumer, consumer.endOffsets(partitions), withRepartitioning, false);
          } catch (final Exception e) {
              e.printStackTrace(System.err);
              System.out.println("FAILED");
              return;
          }
  
          verifyReceivedAllRecords(inputRecordsPerTopicPerPartition.get("data"), outputRecordsPerTopicPerPartition.get("echo"));
          if (withRepartitioning) {
              verifyReceivedAllRecords(inputRecordsPerTopicPerPartition.get("data"), outputRecordsPerTopicPerPartition.get("repartition"));
          }
  
          verifyMin(inputRecordsPerTopicPerPartition.get("data"), outputRecordsPerTopicPerPartition.get("min"));
          verifySum(inputRecordsPerTopicPerPartition.get("data"), outputRecordsPerTopicPerPartition.get("sum"));
  
          if (withRepartitioning) {
              verifyMax(inputRecordsPerTopicPerPartition.get("repartition"), outputRecordsPerTopicPerPartition.get("max"));
              verifyCnt(inputRecordsPerTopicPerPartition.get("repartition"), outputRecordsPerTopicPerPartition.get("cnt"));
          }
  
          // do not modify: required test output
          System.out.println("ALL-RECORDS-DELIVERED");
          System.out.flush();
      }
  
      private static void ensureStreamsApplicationDown(final Admin adminClient, final String groupProtocol) {
          final long maxWaitTime = System.currentTimeMillis() + MAX_IDLE_TIME_MS;
          boolean isEmpty;
          do {
              if (Objects.equals(groupProtocol, "streams")) {
                  final StreamsGroupDescription description = getStreamsGroupDescription(adminClient);
                  isEmpty = description.members().isEmpty();
                  if (System.currentTimeMillis() > maxWaitTime && !isEmpty) {
                      throwNotDownException(description);
                  }
              } else {
                  final ConsumerGroupDescription description = getConsumerGroupDescription(adminClient);
                  isEmpty = description.members().isEmpty();
                  if (System.currentTimeMillis() > maxWaitTime && !isEmpty) {
                      throwNotDownException(description);
                  }
              }
              sleep(1000L);
          } while (!isEmpty);
      }
  
      private static void throwNotDownException(final Object description) {
          throw new RuntimeException(
              "Streams application not down after " + MAX_IDLE_TIME_MS / 1000L + " seconds. " +
                  "Group: " + description
          );
      }
  
      private static Map<TopicPartition, Long> getCommittedOffsets(final Admin adminClient,
                                                                   final boolean withRepartitioning) {
          final Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap;
  
          try {
              final ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult = adminClient.listConsumerGroupOffsets(EosTestClient.APP_ID);
              topicPartitionOffsetAndMetadataMap = listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().get(10, TimeUnit.SECONDS);
          } catch (final Exception e) {
              e.printStackTrace();
              throw new RuntimeException(e);
          }
  
          final Map<TopicPartition, Long> committedOffsets = new HashMap<>();
  
          for (final Map.Entry<TopicPartition, OffsetAndMetadata> entry : topicPartitionOffsetAndMetadataMap.entrySet()) {
              final String topic = entry.getKey().topic();
              if (topic.equals("data") || withRepartitioning && topic.equals("repartition")) {
                  committedOffsets.put(entry.getKey(), entry.getValue().offset());
              }
          }
  
          return committedOffsets;
      }
  
      private static Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> getRecords(final KafkaConsumer<byte[], byte[]> consumer,
                                                                                                       final Map<TopicPartition, Long> readEndOffsets,
                                                                                                       final boolean withRepartitioning,
                                                                                                       final boolean isInputTopic) {
          System.out.println("read end offset: " + readEndOffsets);
          final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> recordPerTopicPerPartition = new HashMap<>();
          final Map<TopicPartition, Long> maxReceivedOffsetPerPartition = new HashMap<>();
          final Map<TopicPartition, Long> maxConsumerPositionPerPartition = new HashMap<>();
  
          long maxWaitTime = System.currentTimeMillis() + MAX_IDLE_TIME_MS;
          boolean allRecordsReceived = false;
          while (!allRecordsReceived && System.currentTimeMillis() < maxWaitTime) {
              final ConsumerRecords<byte[], byte[]> receivedRecords = consumer.poll(Duration.ofSeconds(1L));
  
              for (final ConsumerRecord<byte[], byte[]> record : receivedRecords) {
                  maxWaitTime = System.currentTimeMillis() + MAX_IDLE_TIME_MS;
                  final TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                  maxReceivedOffsetPerPartition.put(tp, record.offset());
                  final long readEndOffset = readEndOffsets.get(tp);
                  if (record.offset() < readEndOffset) {
                      addRecord(record, recordPerTopicPerPartition, withRepartitioning);
                  } else if (!isInputTopic) {
                      throw new RuntimeException("FAIL: did receive more records than expected for " + tp
                          + " (expected EOL offset: " + readEndOffset + "; current offset: " + record.offset());
                  }
              }
  
              for (final TopicPartition tp : readEndOffsets.keySet()) {
                  maxConsumerPositionPerPartition.put(tp, consumer.position(tp));
                  if (consumer.position(tp) >= readEndOffsets.get(tp)) {
                      consumer.pause(Collections.singletonList(tp));
                  }
              }
  
              allRecordsReceived = consumer.paused().size() == readEndOffsets.keySet().size();
          }
  
          if (!allRecordsReceived) {
              System.err.println("Pause partitions (ie, received all data): " + consumer.paused());
              System.err.println("Max received offset per partition: " + maxReceivedOffsetPerPartition);
              System.err.println("Max consumer position per partition: " + maxConsumerPositionPerPartition);
              throw new RuntimeException("FAIL: did not receive all records after " + (MAX_IDLE_TIME_MS / 1000L) + " sec idle time.");
          }
  
          return recordPerTopicPerPartition;
      }
  
      private static void addRecord(final ConsumerRecord<byte[], byte[]> record,
                                    final Map<String, Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>>> recordPerTopicPerPartition,
                                    final boolean withRepartitioning) {
  
          final String topic = record.topic();
          final TopicPartition partition = new TopicPartition(topic, record.partition());
  
          if (verifyTopic(topic, withRepartitioning)) {
              final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> topicRecordsPerPartition =
                  recordPerTopicPerPartition.computeIfAbsent(topic, k -> new HashMap<>());
  
              final List<ConsumerRecord<byte[], byte[]>> records =
                  topicRecordsPerPartition.computeIfAbsent(partition, k -> new ArrayList<>());
  
              records.add(record);
          } else {
              throw new RuntimeException("FAIL: received data from unexpected topic: " + record);
          }
      }
  
      private static boolean verifyTopic(final String topic,
                                         final boolean withRepartitioning) {
          final boolean validTopic = "data".equals(topic) || "echo".equals(topic) || "min".equals(topic) || "sum".equals(topic);
  
          if (withRepartitioning) {
              return validTopic || "repartition".equals(topic) || "max".equals(topic) || "cnt".equals(topic);
          }
  
          return validTopic;
      }
  
      private static void verifyReceivedAllRecords(final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> expectedRecords,
                                                   final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> receivedRecords) {
          if (expectedRecords.size() != receivedRecords.size()) {
              throw new RuntimeException("Result verification failed. Received " + receivedRecords.size() + " records but expected " + expectedRecords.size());
          }
  
          final StringDeserializer stringDeserializer = new StringDeserializer();
          final IntegerDeserializer integerDeserializer = new IntegerDeserializer();
          for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : receivedRecords.entrySet()) {
              final TopicPartition inputTopicPartition = new TopicPartition("data", partitionRecords.getKey().partition());
              final List<ConsumerRecord<byte[], byte[]>> receivedRecordsForPartition = partitionRecords.getValue();
              final List<ConsumerRecord<byte[], byte[]>> expectedRecordsForPartition = expectedRecords.get(inputTopicPartition);
  
              System.out.println(partitionRecords.getKey() + " with " + receivedRecordsForPartition.size() + ", " +
                      inputTopicPartition + " with " + expectedRecordsForPartition.size());
  
              final Iterator<ConsumerRecord<byte[], byte[]>> expectedRecord = expectedRecordsForPartition.iterator();
              RuntimeException exception = null;
              for (final ConsumerRecord<byte[], byte[]> receivedRecord : receivedRecordsForPartition) {
                  if (!expectedRecord.hasNext()) {
                      exception = new RuntimeException("Result verification failed for " + receivedRecord + " since there's no more expected record");
                  }
  
                  final ConsumerRecord<byte[], byte[]> expected = expectedRecord.next();
  
                  final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
                  final int receivedValue = integerDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
                  final String expectedKey = stringDeserializer.deserialize(expected.topic(), expected.key());
                  final int expectedValue = integerDeserializer.deserialize(expected.topic(), expected.value());
  
                  if (!receivedKey.equals(expectedKey) || receivedValue != expectedValue) {
                      exception = new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + expectedKey + "," + expectedValue + "> but was <" + receivedKey + "," + receivedValue + ">");
                  }
              }
  
              if (exception != null) {
                  throw exception;
              }
          }
      }
  
      private static void verifyMin(final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> inputPerTopicPerPartition,
                                    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> minPerTopicPerPartition) {
          final StringDeserializer stringDeserializer = new StringDeserializer();
          final IntegerDeserializer integerDeserializer = new IntegerDeserializer();
  
          final HashMap<String, Integer> currentMinPerKey = new HashMap<>();
          for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : minPerTopicPerPartition.entrySet()) {
              final TopicPartition inputTopicPartition = new TopicPartition("data", partitionRecords.getKey().partition());
              final List<ConsumerRecord<byte[], byte[]>> partitionInput = inputPerTopicPerPartition.get(inputTopicPartition);
              final List<ConsumerRecord<byte[], byte[]>> partitionMin = partitionRecords.getValue();
  
              if (partitionInput.size() != partitionMin.size()) {
                  throw new RuntimeException("Result verification failed: expected " + partitionInput.size() + " records for "
                      + partitionRecords.getKey() + " but received " + partitionMin.size());
              }
  
              final Iterator<ConsumerRecord<byte[], byte[]>> inputRecords = partitionInput.iterator();
  
              for (final ConsumerRecord<byte[], byte[]> receivedRecord : partitionMin) {
                  final ConsumerRecord<byte[], byte[]> input = inputRecords.next();
  
                  final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
                  final int receivedValue = integerDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
                  final String key = stringDeserializer.deserialize(input.topic(), input.key());
                  final int value = integerDeserializer.deserialize(input.topic(), input.value());
  
                  Integer min = currentMinPerKey.get(key);
                  if (min == null) {
                      min = value;
                  } else {
                      min = Math.min(min, value);
                  }
                  currentMinPerKey.put(key, min);
  
                  if (!receivedKey.equals(key) || receivedValue != min) {
                      throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + min + "> but was <" + receivedKey + "," + receivedValue + ">");
                  }
              }
          }
      }
  
      private static void verifySum(final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> inputPerTopicPerPartition,
                                    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> minPerTopicPerPartition) {
          final StringDeserializer stringDeserializer = new StringDeserializer();
          final IntegerDeserializer integerDeserializer = new IntegerDeserializer();
          final LongDeserializer longDeserializer = new LongDeserializer();
  
          final HashMap<String, Long> currentSumPerKey = new HashMap<>();
          for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : minPerTopicPerPartition.entrySet()) {
              final TopicPartition inputTopicPartition = new TopicPartition("data", partitionRecords.getKey().partition());
              final List<ConsumerRecord<byte[], byte[]>> partitionInput = inputPerTopicPerPartition.get(inputTopicPartition);
              final List<ConsumerRecord<byte[], byte[]>> partitionSum = partitionRecords.getValue();
  
              if (partitionInput.size() != partitionSum.size()) {
                  throw new RuntimeException("Result verification failed: expected " + partitionInput.size() + " records for "
                      + partitionRecords.getKey() + " but received " + partitionSum.size());
              }
  
              final Iterator<ConsumerRecord<byte[], byte[]>> inputRecords = partitionInput.iterator();
  
              for (final ConsumerRecord<byte[], byte[]> receivedRecord : partitionSum) {
                  final ConsumerRecord<byte[], byte[]> input = inputRecords.next();
  
                  final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
                  final long receivedValue = longDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
                  final String key = stringDeserializer.deserialize(input.topic(), input.key());
                  final int value = integerDeserializer.deserialize(input.topic(), input.value());
  
                  Long sum = currentSumPerKey.get(key);
                  if (sum == null) {
                      sum = (long) value;
                  } else {
                      sum += value;
                  }
                  currentSumPerKey.put(key, sum);
  
                  if (!receivedKey.equals(key) || receivedValue != sum) {
                      throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + sum + "> but was <" + receivedKey + "," + receivedValue + ">");
                  }
              }
          }
      }
  
      private static void verifyMax(final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> inputPerTopicPerPartition,
                                    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> maxPerTopicPerPartition) {
          final StringDeserializer stringDeserializer = new StringDeserializer();
          final IntegerDeserializer integerDeserializer = new IntegerDeserializer();
  
          final HashMap<String, Integer> currentMinPerKey = new HashMap<>();
          for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : maxPerTopicPerPartition.entrySet()) {
              final TopicPartition inputTopicPartition = new TopicPartition("repartition", partitionRecords.getKey().partition());
              final List<ConsumerRecord<byte[], byte[]>> partitionInput = inputPerTopicPerPartition.get(inputTopicPartition);
              final List<ConsumerRecord<byte[], byte[]>> partitionMax = partitionRecords.getValue();
  
              if (partitionInput.size() != partitionMax.size()) {
                  throw new RuntimeException("Result verification failed: expected " + partitionInput.size() + " records for "
                      + partitionRecords.getKey() + " but received " + partitionMax.size());
              }
  
              final Iterator<ConsumerRecord<byte[], byte[]>> inputRecords = partitionInput.iterator();
  
              for (final ConsumerRecord<byte[], byte[]> receivedRecord : partitionMax) {
                  final ConsumerRecord<byte[], byte[]> input = inputRecords.next();
  
                  final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
                  final int receivedValue = integerDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
                  final String key = stringDeserializer.deserialize(input.topic(), input.key());
                  final int value = integerDeserializer.deserialize(input.topic(), input.value());
  
  
                  Integer max = currentMinPerKey.get(key);
                  if (max == null) {
                      max = Integer.MIN_VALUE;
                  }
                  max = Math.max(max, value);
                  currentMinPerKey.put(key, max);
  
                  if (!receivedKey.equals(key) || receivedValue != max) {
                      throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + max + "> but was <" + receivedKey + "," + receivedValue + ">");
                  }
              }
          }
      }
  
      private static void verifyCnt(final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> inputPerTopicPerPartition,
                                    final Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> cntPerTopicPerPartition) {
          final StringDeserializer stringDeserializer = new StringDeserializer();
          final LongDeserializer longDeserializer = new LongDeserializer();
  
          final HashMap<String, Long> currentSumPerKey = new HashMap<>();
          for (final Map.Entry<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> partitionRecords : cntPerTopicPerPartition.entrySet()) {
              final TopicPartition inputTopicPartition = new TopicPartition("repartition", partitionRecords.getKey().partition());
              final List<ConsumerRecord<byte[], byte[]>> partitionInput = inputPerTopicPerPartition.get(inputTopicPartition);
              final List<ConsumerRecord<byte[], byte[]>> partitionCnt = partitionRecords.getValue();
  
              if (partitionInput.size() != partitionCnt.size()) {
                  throw new RuntimeException("Result verification failed: expected " + partitionInput.size() + " records for "
                      + partitionRecords.getKey() + " but received " + partitionCnt.size());
              }
  
              final Iterator<ConsumerRecord<byte[], byte[]>> inputRecords = partitionInput.iterator();
  
              for (final ConsumerRecord<byte[], byte[]> receivedRecord : partitionCnt) {
                  final ConsumerRecord<byte[], byte[]> input = inputRecords.next();
  
                  final String receivedKey = stringDeserializer.deserialize(receivedRecord.topic(), receivedRecord.key());
                  final long receivedValue = longDeserializer.deserialize(receivedRecord.topic(), receivedRecord.value());
                  final String key = stringDeserializer.deserialize(input.topic(), input.key());
  
                  Long cnt = currentSumPerKey.get(key);
                  if (cnt == null) {
                      cnt = 0L;
                  }
                  currentSumPerKey.put(key, ++cnt);
  
                  if (!receivedKey.equals(key) || receivedValue != cnt) {
                      throw new RuntimeException("Result verification failed for " + receivedRecord + " expected <" + key + "," + cnt + "> but was <" + receivedKey + "," + receivedValue + ">");
                  }
              }
          }
      }
  
      private static void verifyAllTransactionFinished(final KafkaConsumer<byte[], byte[]> consumer,
                                                       final String kafka,
                                                       final boolean withRepartitioning) {
          final String[] topics;
          if (withRepartitioning) {
              topics = new String[] {"echo", "min", "sum", "repartition", "max", "cnt"};
          } else {
              topics = new String[] {"echo", "min", "sum"};
          }
  
          final List<TopicPartition> partitions = getAllPartitions(consumer, topics);
          consumer.assign(partitions);
          consumer.seekToEnd(partitions);
          for (final TopicPartition tp : partitions) {
              System.out.println(tp + " at position " + consumer.position(tp));
          }
  
          final Properties consumerProps = new Properties();
          consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumer-uncommitted");
          consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
          consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
          consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
  
  
          final long maxWaitTime = System.currentTimeMillis() + MAX_IDLE_TIME_MS;
          try (final KafkaConsumer<byte[], byte[]> consumerUncommitted = new KafkaConsumer<>(consumerProps)) {
              while (!partitions.isEmpty() && System.currentTimeMillis() < maxWaitTime) {
                  consumer.seekToEnd(partitions);
                  final Map<TopicPartition, Long> topicEndOffsets = consumerUncommitted.endOffsets(partitions);
  
                  final Iterator<TopicPartition> iterator = partitions.iterator();
                  while (iterator.hasNext()) {
                      final TopicPartition topicPartition = iterator.next();
                      final long position = consumer.position(topicPartition);
  
                      if (position == topicEndOffsets.get(topicPartition)) {
                          iterator.remove();
                          System.out.println("Removing " + topicPartition + " at position " + position);
                      } else if (consumer.position(topicPartition) > topicEndOffsets.get(topicPartition)) {
                          throw new IllegalStateException("Offset for partition " + topicPartition + " is larger than topic endOffset: " + position + " > " + topicEndOffsets.get(topicPartition));
                      } else {
                          System.out.println("Retry " + topicPartition + " at position " + position);
                      }
                  }
                  sleep(1000L);
              }
          }
  
          if (!partitions.isEmpty()) {
              throw new RuntimeException("Could not read all verification records. Did not receive any new record within the last " + (MAX_IDLE_TIME_MS / 1000L) + " sec.");
          }
      }
  
      private static List<TopicPartition> getAllPartitions(final KafkaConsumer<?, ?> consumer,
                                                           final String... topics) {
          final ArrayList<TopicPartition> partitions = new ArrayList<>();
  
          for (final String topic : topics) {
              for (final PartitionInfo info : consumer.partitionsFor(topic)) {
                  partitions.add(new TopicPartition(info.topic(), info.partition()));
              }
          }
          return partitions;
      }
  
      private static ConsumerGroupDescription getConsumerGroupDescription(final Admin adminClient) {
          final ConsumerGroupDescription description;
          try {
              description = adminClient.describeConsumerGroups(Collections.singleton(EosTestClient.APP_ID))
                  .describedGroups()
                  .get(EosTestClient.APP_ID)
                  .get(10, TimeUnit.SECONDS);
          } catch (final InterruptedException | ExecutionException | java.util.concurrent.TimeoutException e) {
              e.printStackTrace();
              throw new RuntimeException("Unexpected Exception getting group description", e);
          }
          return description;
      }
  
      private static StreamsGroupDescription getStreamsGroupDescription(final Admin adminClient) {
          final StreamsGroupDescription description;
          try {
              description = adminClient.describeStreamsGroups(Collections.singleton(EosTestClient.APP_ID))
                  .describedGroups()
                  .get(EosTestClient.APP_ID)
                  .get(10, TimeUnit.SECONDS);
          } catch (final InterruptedException | ExecutionException | java.util.concurrent.TimeoutException e) {
              e.printStackTrace();
              throw new RuntimeException("Unexpected Exception getting group description", e);
          }
          return description;
      }
  }
  