THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
This is for adding mocks for state stores used in Streams unit testing. We'd like to use mocks for different types of state stores: KV, window, session - that can be used to record the number of expected put / get calls used in the DSL operator unit testing. These will provide convenience for developers when they are writing unit test for Kafka stream and other related modules.
For example, in the current streams/TopologyTest.java:
| Code Block | ||
|---|---|---|
| ||
...
@Test
public void shouldFailIfSinkIsParent() {
topology.addSource("source", "topic-1");
topology.addSink("sink-1", "topic-2", "source");
try {
topology.addSink("sink-2", "topic-3", "sink-1");
fail("Should throw TopologyException for using sink as parent");
} catch (final TopologyException expected) { }
}
@Test(expected = TopologyException.class)
public void shouldNotAllowToAddStateStoreToNonExistingProcessor() {
mockStoreBuilder();
EasyMock.replay(storeBuilder);
topology.addStateStore(storeBuilder, "no-such-processor");
}
@Test
public void shouldNotAllowToAddStateStoreToSource() {
mockStoreBuilder();
EasyMock.replay(storeBuilder);
topology.addSource("source-1", "topic-1");
try {
topology.addStateStore(storeBuilder, "source-1");
fail("Should have thrown TopologyException for adding store to source node");
} catch (final TopologyException expected) { }
}
private void mockStoreBuilder() {
EasyMock.expect(storeBuilder.name()).andReturn("store").anyTimes();
EasyMock.expect(storeBuilder.logConfig()).andReturn(Collections.emptyMap());
EasyMock.expect(storeBuilder.loggingEnabled()).andReturn(false);
} |
One of the goal is to replace the in-test vanilla mockStoreBuilder with a more generic mockStoreBuilder class, which can provide more functionality and can be reuse later.
Public Interfaces
We add some new classes to the kafka-streams-test-utils under a new org.apache.kafka.streams.state package.
...