From 600b0b10f8e98bebef80f75e391a78c459ffb45c Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Wed, 17 Aug 2022 22:31:19 +0200 Subject: [PATCH] =?utf8?q?ROT:=20(Ohne=20stored-offsets)=20=C3=9Cberpr?= =?utf8?q?=C3=BCfung=20der=20Fachlogik=20korrigiert?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit * Der ursprüungliche Test ist nicht korrekt angeschlangen * Der Test Schlug nicht an, weil geprüft wurde, dass `AdderResults` eine Teilmenge der insgesamt erwarteten Ergebnisse enthält, aber nicht mehr und/oder andere Ergebnisse. * Problem: `AdderResult` hat zum Zeitpunkt der Überprüfung überhaupt keine Ergebnisse enthalten, da der Consumer nach dem Fehler alle Partitionen abgegeben hat und entsprechend die Ergebnisse aus `AdderResult` entfernt und gespeichert wurden. * Daher wird jetzt gegen die in der Mongo-DB gespeicherten Ergebnisse verglichen. * Unterwegs verbessert / korrigiert: ** Falsches Assert-Statement entfernt (beim 2. Durchlauf können durchaus mehr Nachrichten als erwartet empfangen werden, nämlich 2x weniger als erwartet ;) ** Commit erfolgt alle 500ms ** Test realistischer gestaltet: Viel mehr Nachrichten und durcheinander. ** Der Fehler wird nicht nach der ersten Hand voll Nachrichten erzeugt, sondern erst gegen Ende der generierten Nachrichten. --- .../java/de/juplo/kafka/ApplicationIT.java | 2 +- .../java/de/juplo/kafka/ApplicationTests.java | 84 +++++++++++-------- .../juplo/kafka/GenericApplicationTests.java | 5 +- 3 files changed, 52 insertions(+), 39 deletions(-) diff --git a/src/test/java/de/juplo/kafka/ApplicationIT.java b/src/test/java/de/juplo/kafka/ApplicationIT.java index d1d8e50..cded0ee 100644 --- a/src/test/java/de/juplo/kafka/ApplicationIT.java +++ b/src/test/java/de/juplo/kafka/ApplicationIT.java @@ -32,7 +32,7 @@ public class ApplicationIT @Test - public void testApplicationStartup() + public void testApplicationStartup() { restTemplate.getForObject( "http://localhost:" + port + "/actuator/health", diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index 1336050..740c09c 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -1,5 +1,6 @@ package de.juplo.kafka; +import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.Bytes; @@ -13,10 +14,11 @@ import java.util.stream.IntStream; import static org.assertj.core.api.Assertions.assertThat; +@Slf4j public class ApplicationTests extends GenericApplicationTests { @Autowired - AdderResults results; + StateRepository stateRepository; public ApplicationTests() @@ -30,7 +32,7 @@ public class ApplicationTests extends GenericApplicationTests { ApplicationTests tests; - final int[] numbers = {1, 7, 3, 2, 33, 6, 11}; + final int[] numbers = {1, 77, 33, 2, 66, 666, 11}; final String[] dieWilden13 = IntStream .range(1, 14) @@ -57,40 +59,51 @@ public class ApplicationTests extends GenericApplicationTests 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)); + int number[] = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 }; + int message[] = { 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1 }; + int next = 0; - for (int message = 1; message <= number; message++) + for (int pass = 0; pass < 333; pass++) + { + for (int i = 0; i<13; i++) { - Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message))); - send(key, value, logicErrors, messageSender); - } - send(key, calculateMessage, logicErrors, messageSender); + String seeräuber = dieWilden13[i]; + Bytes key = new Bytes(stringSerializer.serialize(TOPIC, seeräuber)); - state.get(seeräuber).add(new AdderResult(number, (number + 1) * number / 2)); + if (message[i] > number[i]) + { + send(key, calculateMessage, fail(logicErrors, pass, counter), messageSender); + state.get(seeräuber).add(new AdderResult(number[i], (number[i] + 1) * number[i] / 2)); + // Pick next number to calculate + number[i] = numbers[next++%numbers.length]; + message[i] = 1; + log.debug("Seeräuber {} will die Summe für {} berechnen", seeräuber, number[i]); + } + + Bytes value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(message[i]++))); + send(key, value, fail(logicErrors, pass, counter), messageSender); + } } return counter; } + boolean fail (boolean logicErrors, int pass, int counter) + { + return logicErrors && pass > 300 && counter%77 == 0; + } + void send( Bytes key, Bytes value, - boolean logicErrors, + boolean fail, Consumer> messageSender) { counter++; - if (counter == 77) + if (fail) { - if (logicErrors) - { - value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(-1))); - } + value = new Bytes(stringSerializer.serialize(TOPIC, Integer.toString(-1))); } messageSender.accept(new ProducerRecord<>(TOPIC, key, value)); @@ -105,20 +118,23 @@ public class ApplicationTests extends GenericApplicationTests @Override public void assertBusinessLogic() { - tests.results - .getState() - .values() - .stream() - .flatMap(map -> map.entrySet().stream()) - .forEach(entry -> - { - String user = entry.getKey(); - List resultsForUser = entry.getValue(); - - assertThat(state.get(user)) - .describedAs("Unexpected results for user {}", user) - .containsExactlyElementsOf(resultsForUser); - }); + for (int i=0; i + { + String user = entry.getKey(); + List resultsForUser = entry.getValue(); + + assertThat(state.get(user)) + .as("Unexpected results for user %s", user) + .containsExactlyElementsOf(resultsForUser); + }); + } } } } diff --git a/src/test/java/de/juplo/kafka/GenericApplicationTests.java b/src/test/java/de/juplo/kafka/GenericApplicationTests.java index 449c389..93daf6b 100644 --- a/src/test/java/de/juplo/kafka/GenericApplicationTests.java +++ b/src/test/java/de/juplo/kafka/GenericApplicationTests.java @@ -41,7 +41,7 @@ import static org.awaitility.Awaitility.*; properties = { "sumup.adder.bootstrap-server=${spring.embedded.kafka.brokers}", "sumup.adder.topic=" + TOPIC, - "sumup.adder.commit-interval=1s", + "sumup.adder.commit-interval=500ms", "spring.mongodb.embedded.version=4.4.13" }) @EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS) @EnableAutoConfiguration @@ -177,9 +177,6 @@ abstract class GenericApplicationTests checkSeenOffsetsForProgress(); compareToCommitedOffsets(oldOffsets); - assertThat(receivedRecords.size()) - .describedAs("Received not all sent events") - .isLessThan(numberOfGeneratedMessages); assertThatNoException() .describedAs("Consumer should not be running") -- 2.20.1