+ ApplicationTests tests;
+
+ final int[] numbers = {1, 7, 3, 2, 33, 6, 11};
+ final String[] dieWilden13 =
+ IntStream
+ .range(1, 14)
+ .mapToObj(i -> "seeräuber-" + i)
+ .toArray(i -> new String[i]);
+ final StringSerializer stringSerializer = new StringSerializer();
+ final Bytes calculateMessage = new Bytes(stringSerializer.serialize(TOPIC, "CALCULATE"));
+
+ int counter = 0;
+
+ Map<String, List<AdderResult>> state;
+
+ @Override
+ public int generate(
+ boolean poisonPills,
+ boolean logicErrors,
+ Consumer<ProducerRecord<Bytes, Bytes>> messageSender)
+ {
+ counter = 0;
+ state =
+ Arrays
+ .stream(dieWilden13)
+ .collect(Collectors.toMap(
+ seeräuber -> seeräuber,
+ seeräuber -> new LinkedList()));
+
+ for (int i = 0; i < 33; i++)
+ {
+ String seeräuber = dieWilden13[i % 13];
+ int number = numbers[i % 7];
+
+ Bytes key = new Bytes(stringSerializer.serialize(TOPIC, seeräuber));
+
+ for (int message = 1; message <= number; message++)
+ {
+ Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message)));
+ send(key, value, logicErrors, messageSender);
+ }
+ send(key, calculateMessage, logicErrors, messageSender);
+
+ state.get(seeräuber).add(new AdderResult(number, (number + 1) * number / 2));
+ }
+
+ return counter;
+ }
+
+ void send(
+ Bytes key,
+ Bytes value,
+ boolean logicErrors,
+ Consumer<ProducerRecord<Bytes, Bytes>> messageSender)