Fix.

transferwise · Dec 15, 2023 · f6a7cfc · f6a7cfc
1 parent dc041a7
commit f6a7cfc
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java b/tw-tkms-starter/src/test/java/com/transferwise/kafka/tkms/EndToEndIntTest.java
@@ -251,13 +251,19 @@ void testExactlyOnceDelivery(int scenario) throws Exception {
   void testThatMessagesWithSameKeyEndUpInOnePartition(boolean deferUntilCommit) {
     setupConfig(deferUntilCommit);
 
-    String message = "Hello World!";
+    String protoMessage = "Hello Estonia!";
+    String message = StringUtils.repeat(protoMessage, 100);
     String key = "GrailsRocks";
-    int n = 20;
+    int n = 200;
     ConcurrentHashMap<Integer, AtomicInteger> partitionsMap = new ConcurrentHashMap<>();
     AtomicInteger receivedCount = new AtomicInteger();
 
     Consumer<ConsumerRecord<String, String>> messageCounter = cr -> {
+      var testEvent = ExceptionUtils.doUnchecked(() -> objectMapper.readValue(cr.value(), TestEvent.class));
+      if (!message.equals(testEvent.getMessage())) {
+        throw new IllegalStateException("Unexpected message '" + message + "' received.");
+      }
+
       partitionsMap.computeIfAbsent(cr.partition(), (k) -> new AtomicInteger()).incrementAndGet();
       receivedCount.incrementAndGet();
     };