private static <K, V> void compareKeyValueTimestamp(final ConsumerRecord<K, V> record,
                                                        final K expectedKey,
                                                        final V expectedValue,
                                                        final long expectedTimestamp) {
        Objects.requireNonNull(record);
        final K recordKey = record.key();
        final V recordValue = record.value();
        final long recordTimestamp = record.timestamp();
        final AssertionError error = new AssertionError(
            "Expected <" + expectedKey + ", " + expectedValue + "> with timestamp=" + expectedTimestamp +
                " but was <" + recordKey + ", " + recordValue + "> with timestamp=" + recordTimestamp
        );
        if (recordKey != null) {
            if (!recordKey.equals(expectedKey)) {
                throw error;
            }
        } else if (expectedKey != null) {
            throw error;
        }
        if (recordValue != null) {
            if (!recordValue.equals(expectedValue)) {
                throw error;
            }
        } else if (expectedValue != null) {
            throw error;
        }
        if (recordTimestamp != expectedTimestamp) {
            throw error;
        }
    }
Waits for the given {@link KafkaStreams} instances to all be in a specific {@link State}.
Prefer {@link #startApplicationAndWaitUntilRunning(List, Duration)} when possible
because this method uses polling, which can be more error prone and slightly slower.

@param streamsList the list of streams instances to run.
@param state the expected state that all the streams to be in within timeout
@param timeout the time to wait for the streams to all be in the specific state.

@throws InterruptedException if the streams doesn't change to the expected state in time.
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java