Versions Compared

Old Version 9

changes.mady.by.user Jukka Karvanen

Saved on

compared with

New Version 10

changes.mady.by.user Jukka Karvanen

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Removed unmodified methods, Instant and Duration to use

...

Code Block
languagejava
titleTopologyTestDriver
package org.apache.kafka.streams;

public class TopologyTestDriver {
	public TopologyTestDriver(Topology topology, Properties config); // initialized WallClockMockTimeMs with System.currentTimeMillis()

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

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

	// event-time is automatically 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
	public final <K, V> TestInputTopic<KTestOutputTopic<K, V> createInputTopiccreateOutputTopic(final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde);
	// Uses current system time as start timestamp. Auto-advance is disabled.
	public final <K, V> TestOutputTopic<KTestInputTopic<K, V> createOutputTopiccreateInputTopic(final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde);

	public Map<String, StateStore> getAllStateStores()
	public StateStore getStateStore(String name);    //Uses provided startTimestamp and autoAdvance duration for timestamp generation
	public final <K, V> KeyValueStore<KTestInputTopic<K, V> getKeyValueStorecreateInputTopic(final String name);
	public <K, V> WindowStore<K, V> getWindowStore(String name);
	public <K, V> SessionStore<K, V> getSessionStore(String name);

	public void close();topicName, final Serde<K> keySerde, final Serde<V> valueSerde, final Instant startTimestamp, final Duration autoAdvance);

    ...
}



Code Block
languagejava
titleTestInputTopic
package org.apache.kafka.streams;

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

    //Timestamp handling 
    //Record timestamp can be provided when piping input or use internally tracked time configured with configureTimingparameters:
    //startTimestampMsstartTimestamp the initial timestamp for generated records, if not provided uses current system time as start timestamp.
    //autoAdvanceMsautoAdvance the time increment per generated record, if not provided auto-advance is disabled.

	//Reinitialize timestamp and advanceMs
    public void configureTiming(final long startTimestampMs, final long autoAdvanceMs);
    //Advances the internally tracked time.
    void advanceTimeMs(final longDuration advanceMsadvance);

	//Methods to pipe single record
    void pipeInput(final V value);
    void pipeInput(final K key, final V value);

    // Use provided timestamp, does not auto advance internally tracked time.
    void pipeInput(final V value, final long timestampMs);
    void pipeInput(final K key, final V value, final long timestampMs);

    // If record timestamp set, does not auto advance internally tracked time.
    void pipeInput(final TestRecord<K, V> record);

	//Methods to pipe list of records
    void pipeValueList(final List<V> values);
    void pipeKeyValueList(final List<KeyValue<K, V>> keyValues);

    // Use provided timestamp, does not auto advance internally tracked time.
    void pipeValueList(final List<V> values, final longInstant startTimestamp, final longDuration advanceMs);
    void pipeKeyValueList(final List<KeyValue<K, V>> keyValues, final longInstant startTimestamp, final longDuration advanceMs);

    // If record timestamp set, does not auto advance internally tracked time.
    void pipeRecordList(final List<? extends TestRecord<K, V>> records);
}

...