Mockafka
Mockafka is a DSL which allows testing of kafka-streams topologies without need of a kafka and zookeeper installation. It is a Java 8 port of jpzk/mockedstreams.
Add it to your pom.xml...
<dependency>
<groupId>com.github.carlosmenezes</groupId>
<artifactId>mockafka</artifactId>
<version>0.1.2</version>
<scope>test</scope>
</dependency>or to your build.gradle.
testCompile "com.github.carlosmenezes:mockafka:0.1.2"Examples
List<Message<Integer, Integer>> input = Stream.of(1, 2, 3, 4, 5, 6, 7)
.map(i -> new Message<>(i, i))
.collect(Collectors.toList());
Serde<Integer> integerSerde = Serdes.Integer();
List<Message<Integer, Integer>> output = Mockafka
.builder()
.topology(builder ->
builder.stream(integerSerde, integerSerde, "numbersTopic")
.filter((key, value) -> value % 2 == 1)
.to(integerSerde, integerSerde, "oddNumbersTopic")
)
.input("numbersTopic", integerSerde, integerSerde, input)
.output("oddNumbersTopic", integerSerde, integerSerde, 4);
List<Message<Integer, Integer>> expected = Arrays.asList(new Message<>(1, 1), new Message<>(3, 3), new Message<>(5, 5), new Message<>(7, 7));
assertEquals(4, output.size());
assertEquals(expected, output);Multiple inputs/outputs
MockafkaBuilder builder = Mockafka
.builder()
.topology(builder -> {...})
.input("someInput", Serdes.String(), Serdes.String(), someInput)
.input("anotherInput", Serdes.String(), Serdes.String(), anotherInput);
List<Message<String, String>> someOutput = builder.output("someOutput", Serdes.String(), Serdes.String(), 10);
List<Message<String, String>> anotherOutput = builder.output("anotherOutput", Serdes.String(), Serdes.String(), 10);Using State Stores
You can create state stores using .stores(String... stores) method and verify
it's contents with .stateTable(String name) method:
MockafkaBuilder builder = Mockafka
.builder()
.topology(builder -> {...})
.input("someInput", Serdes.String(), Serdes.String(), someInput)
.stores("someStore");
Map<String, String> someStore = builder.stateTable("someStore");
assertEquals(10, someStore.size());Using Window State Stores
When defining a timestamp extractor in .config(Properties config) you can verify the content as
windowed state stores using the method .windowStateTable(String name, K key, long timeFrom, long timeTo).
Properties properties = new Properties();
properties.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TestTimestampExtractor.class.getName());
MockafkaBuilder builder = Mockafka
.builder()
.topology(TopologyUtil::windowStateTopology)
.input(TopologyUtil.WINDOW_TOPIC, TopologyUtil.stringSerde, TopologyUtil.integerSerde, createInputKeyValueForWindow().toArray(new Message[]{}))
.stores(TEST_STORE)
.config(properties);
Map<String, Long> actualSomeKey = builder.windowStateTable(TEST_STORE, "somekey", 0, Long.MAX_VALUE);
Map<Long, Long> expectedSomeKey = new HashMap<>();
expectedSomeKey.put(25L, 1L);
expectedSomeKey.put(42L, 2L);
assertEquals(expectedSomeKey, actualSomeKey);
Map<String, Long> actualAnotherKey = builder.windowStateTable(TEST_STORE, "anotherkey", 0, Long.MAX_VALUE);
Map<Long, Long> expectedAnotherKey = new HashMap<>();
expectedAnotherKey.put(50L, 2L);
expectedAnotherKey.put(90L, 1L);
assertEquals(expectedAnotherKey, actualAnotherKey);Configuring Streams
If you need some custom configuration, just use the method .config(Properties config):
Properties properties = new Properties();
properties.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, TestTimestampExtractor.class.getName());
MockafkaBuilder builder = Mockafka
.builder()
.topology({...})
.input({...})
.stores({...})
.config(properties);