@SuppressWarnings("unchecked")
    private static <K, V, T> List<T> waitUntilFinalKeyValueRecordsReceived(final Properties consumerConfig,
                                                                           final String topic,
                                                                           final List<T> expectedRecords,
                                                                           final long waitTime,
                                                                           final boolean withTimestamp) throws Exception {
        final List<T> accumData = new ArrayList<>();
        try (final Consumer<K, V> consumer = createConsumer(consumerConfig)) {
            final TestCondition valuesRead = () -> {
                final List<T> readData;
                if (withTimestamp) {
                    readData = (List<T>) readKeyValuesWithTimestamp(topic, consumer, waitTime, expectedRecords.size());
                } else {
                    readData = (List<T>) readKeyValues(topic, consumer, waitTime, expectedRecords.size());
                }
                accumData.addAll(readData);

                // filter out all intermediate records we don't want
                final List<T> accumulatedActual = accumData
                    .stream()
                    .filter(expectedRecords::contains)
                    .collect(Collectors.toList());

                // still need to check that for each key, the ordering is expected
                final Map<K, List<T>> finalAccumData = new HashMap<>();
                for (final T kv : accumulatedActual) {
                    final K key = withTimestamp ? ((KeyValueTimestamp<K, V>) kv).key() : ((KeyValue<K, V>) kv).key;
                    final List<T> records = finalAccumData.computeIfAbsent(key, k -> new ArrayList<>());
                    if (!records.contains(kv)) {
                        records.add(kv);
                    }
                }
                final Map<K, List<T>> finalExpected = new HashMap<>();
                for (final T kv : expectedRecords) {
                    final K key = withTimestamp ? ((KeyValueTimestamp<K, V>) kv).key() : ((KeyValue<K, V>) kv).key;
                    final List<T> records = finalExpected.computeIfAbsent(key, k -> new ArrayList<>());
                    if (!records.contains(kv)) {
                        records.add(kv);
                    }
                }

                // returns true only if the remaining records in both lists are the same and in the same order
                // and the last record received matches the last expected record
                return finalAccumData.equals(finalExpected);

            };
            final String conditionDetails = "Did not receive all " + expectedRecords + " records from topic " +
                topic + " (got " + accumData + ")";
            TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails);
        }
        return accumData;
    }
Wait until final key-value mappings have been consumed.
Duplicate records are not considered in the comparison.

@param consumerConfig     Kafka Consumer configuration
@param topic              Kafka topic to consume from
@param expectedRecords    Expected key-value mappings
@param waitTime           Upper bound of waiting time in milliseconds
@param <K>                Key type of the data records
@param <V>                Value type of the data records
@return All the mappings consumed, or null if no records are consumed
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java