package org.apache.kafka.clients.producer;
public class ProducerRecord<K, V> implements ClientRecord<K, V> {
...

package org.apache.kafka.streams;

public class TopologyTestDriver {
	public TopologyTestDriver(Topology topology, Properties configtrigger wall-clock-time punctuation
	public void advanceWallClockTime(Duration advance); // can trigger wall-clock-time punctuation

    //Deprecate old pipe and read methods
    @Deprecate public void pipeInput(ConsumerRecord<byte[], byte[]> record); // initializedcan WallClockMockTimeMstrigger with System.currentTimeMillis()

	public TopologyTestDriver(Topology topology, Properties config, long initialWallClockTimeMs);

    //Deprecate old pipe and read methodsevent-time punctuation
    @Deprecate public void pipeInput(List<ConsumerRecord<byte[], byte[]>> records); // can trigger event-time punctuation
    @Deprecate public void pipeInput(ConsumerRecord<byteProducerRecord<byte[], byte[]> recordreadOutput(String topic); // can trigger event-time punctuation
    @Deprecate public <K, void pipeInput(List<ConsumerRecord<byte[], byte[]>> records); // can trigger event-time punctuation
    @Deprecate public ProducerRecord<byte[], byte[]> readOutput(String topic);
    @Deprecate publicV> ProducerRecord<K, V> readOutput(String topic, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer);

	// methods for TestTopic object creation
	public final <K, V> ProducerRecord<KTestOutputTopic<K, V> readOutputcreateOutputTopic(final String topictopicName, Deserializer<K>final keyDeserializerSerializer<K> keySerializer, final Deserializer<V>Serializer<V> valueDeserializervalueSerializer);

	// event- Uses current system time isas automaticallystart advanced based on the provided input records, and thus event-time punctuation are triggered automatically
	// wall-clock time is mocked and not advanced automatically; user can advance wall-clock time manually, and thus, trigger wall-clock time punctuations manually
	public void advanceWallClockTime(long advanceMs); // can trigger wall-clock-time punctuation

	// methods for TestTopic object creation
	//TODO investigate, is there way to find out serde based on topology
	public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde);
	public final <K, V> TestOutputTopic<K, V> createOutputTopic(final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde);

	public Map<String, StateStore> getAllStateStores()
	public StateStore getStateStore(String name);
	public <K, V> KeyValueStore<K, V> getKeyValueStore(String name);
	public <K, V> WindowStore<K, V> getWindowStore(String name);
	public <K, V> SessionStore<K, V> getSessionStore(String name);

	public void close();
}
timestamp. Auto-advance is disabled.
	public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName, final Deserializer<K> keyDeserializer, final Deserializer<V> valueDeserializer);
    //Uses provided startTimestamp and autoAdvance duration for timestamp generation
	public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName, final Deserializer<K> keyDeserializer, final Deserializer<V> valueDeserializer, final Instant startTimestamp, final Duration autoAdvance);

    ...
}

package org.apache.kafka.streams;

public class TestOutputTopic<K, V> {
    //Create by TopologyTestDriver, Constructors are package private

	//Method to check queue size
    final long getQueueSize();
    final boolean isEmpty();

	//Methods to readOutput, throw NoSuchElement if no record in queue
    V readValue();
    KeyValue<K, V> readKeyValue();
    ClientRecord<K, V> readRecord();

    //Output as collection
    List<V> readValuesToList();
    Map<K, V> readKeyValuesToMap();
    List<KeyValue<K, V>> readKeyValuesToList();
    List<ClientRecord<K, V>> readRecordsToList();
}

package org.apache.kafka.streams.test;
 
//OutputVerifier modified to accept ClientRecord instead of ProducerRecord
public class OutputVerifier {
 
    public static <K,V> void compareValue(ClientRecord<K, V> record, V expectedValue) throws AssertionError;
    public static <K,V> void compareValue(ClientRecord<K, V> record, ClientRecord<K, V> expectedRecord) throws AssertionError;
 
    public static <K,V> void compareKeyValue(ClientRecord<K, V> record, K expectedKey, V expectedValue) throws AssertionError;
    public static <K,V> void compareKeyValue(ClientRecord<K, V> record, ClientRecord<K, V> expectedRecord) throws AssertionError;
 
    public static <K,V> void compareValueTimestamp(ClientRecord<K, V> record, V expectedValue, long expectedTimestamp) throws AssertionError;
    public static <K,V> void compareValueTimestamp(ClientRecord<K, V> record, ClientRecord<K, V> expectedRecord) throws AssertionError;
 
    public static <K,V> void compareKeyValueTimestamp(ClientRecord<K, V> record, K expectedKey, V expectedValue, long expectedTimestamp) throws AssertionError;
    public static <K,V> void compareKeyValueTimestamp(ClientRecord<K, V> record, ClientRecord<K, V> expectedRecord) throws AssertionError;
 
}

package org.apache.kafka.streams.test;

@Deprecated
public class ConsumerRecordFactory<K, V> {
...

package org.apache.kafka.streams.test;
	
//Mainly used by TestInputTopic
public class TestRecordFactory<K, V> {
    // default
    public TestRecordFactory(final String topicName) // initialized startTimestampMs with System.currentTimeMillis() and autoAdvanceMs with zero
    public TestRecordFactory(final String topicName, long startTimestampMs);
    public TestRecordFactory(final String topicName, long startTimestampMs, long autoAdvanceMs);
	
	//Reinitialize timestamp and advanceMs
    public void configureTiming(final long startTimestampMs);
    public void configureTiming(final long startTimestampMs, final long autoAdvanceMs);

	//Advance current timestamp
    public void advanceTimeMs(final long advanceMs);
	
	// create single records with default topic name and given timestamp
    public TestRecord<K, V> create(K key, V value, long timestampMs);
    public TestRecord<K, V> create(V value, long timestampMs);
    public TestRecord<K, V> create(K key, V value, Headers headers, long timestampMs);
    public TestRecord<K, V> create(V value, Headers headers, long timestampMs);

    // create single records with default topic name and generated timestamp
	public TestRecord<K, V> create(K key, V value);
    public TestRecord<K, V> create(V value); 
    public TestRecord<K, V> create(K key, V value, Headers headers);
    public TestRecord<K, V> create(V value, Headers headers);

 
    // create list of records with default topic name internally tracked time
    public List<TestRecord<K, V>> create(final List<KeyValue<K, V>> keyValues);
	// does not auto advance internally tracked time.
    public List<TestRecord<K, V>> create(final List<KeyValue<K, V>> keyValues, long startTimestamp, long advanceMs);
    public List<TestRecord<K, V>> create(final List<KeyValue<K, V>> keyValues, long startTimestamp); //advanceMs =1

}