- compareToCommitedOffsets(newOffsets);
- assertThat(receivedRecords.size())
- .describedAs("Received not all sent events")
- .isLessThan(100);
+ assertThat(endlessConsumer.isRunning())
+ .describedAs("Consumer should still be running")
+ .isTrue();
+ }
+
+ @Test
+ void commitsOffsetOnProgramLogicErrorFoo()
+ {
+ clientMessageHandler.testHandler = (clientMessage, metadata) ->
+ {
+ if (Integer.parseInt(clientMessage.message)%10 ==0)
+ throw new RuntimeException("BOOM: " + clientMessage.message + "%10 == 0");
+ };
+
+ send100Messages((key, counter) -> serialize(key, counter));
+
+ await("80 records received")
+ .atMost(Duration.ofSeconds(30))
+ .until(() -> received.size() == 100);
+
+ await("Offsets committed")
+ .atMost(Duration.ofSeconds(10))
+ .pollDelay(Duration.ofSeconds(1))
+ .untilAsserted(() ->
+ {
+ checkSeenOffsetsForProgress();
+ compareToCommitedOffsets(newOffsets);
+ });
+
+ assertThat(endlessConsumer.isRunning())
+ .describedAs("Consumer should still be running")
+ .isTrue();