X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Ftest%2Fjava%2Fde%2Fjuplo%2Fkafka%2FApplicationTests.java;h=40dc149e519a7f95709d7a7966dddf5b8de722df;hb=0ac94b34af644f6fa5a0556fc7e2bd322167c608;hp=35d13cdda9ecb9749edbaeb6be8fc905fbe30c45;hpb=9d3fa83eb656617162d54a14e1bfcb330fa4ee3a;p=demos%2Fkafka%2Ftraining diff --git a/src/test/java/de/juplo/kafka/ApplicationTests.java b/src/test/java/de/juplo/kafka/ApplicationTests.java index 35d13cd..40dc149 100644 --- a/src/test/java/de/juplo/kafka/ApplicationTests.java +++ b/src/test/java/de/juplo/kafka/ApplicationTests.java @@ -7,10 +7,7 @@ import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.RecordDeserializationException; -import org.apache.kafka.common.serialization.BytesDeserializer; -import org.apache.kafka.common.serialization.BytesSerializer; -import org.apache.kafka.common.serialization.LongSerializer; -import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.common.serialization.*; import org.apache.kafka.common.utils.Bytes; import org.junit.jupiter.api.*; import org.springframework.beans.factory.annotation.Autowired; @@ -34,10 +31,8 @@ import java.util.stream.IntStream; import static de.juplo.kafka.ApplicationTests.PARTITIONS; import static de.juplo.kafka.ApplicationTests.TOPIC; -import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.*; import static org.awaitility.Awaitility.*; -import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; -import static org.junit.jupiter.api.Assertions.assertThrows; @SpringJUnitConfig(initializers = ConfigDataApplicationContextInitializer.class) @@ -55,8 +50,9 @@ class ApplicationTests StringSerializer stringSerializer = new StringSerializer(); - LongSerializer longSerializer = new LongSerializer(); + @Autowired + Serializer valueSerializer; @Autowired KafkaProducer kafkaProducer; @Autowired @@ -81,7 +77,7 @@ class ApplicationTests @Order(1) // << The poistion pill is not skipped. Hence, this test must run first void commitsCurrentOffsetsOnSuccess() throws ExecutionException, InterruptedException { - send100Messages(i -> new Bytes(longSerializer.serialize(TOPIC, i))); + send100Messages(i -> new Bytes(valueSerializer.serialize(TOPIC, i))); await("100 records received") .atMost(Duration.ofSeconds(30)) @@ -95,10 +91,9 @@ class ApplicationTests compareToCommitedOffsets(newOffsets); }); - assertThrows( - IllegalStateException.class, - () -> endlessConsumer.exitStatus(), - "Consumer should still be running"); + assertThatExceptionOfType(IllegalStateException.class) + .isThrownBy(() -> endlessConsumer.exitStatus()) + .describedAs("Consumer should still be running"); } @Test @@ -108,7 +103,7 @@ class ApplicationTests send100Messages(counter -> counter == 77 ? new Bytes(stringSerializer.serialize(TOPIC, "BOOM!")) - : new Bytes(longSerializer.serialize(TOPIC, counter))); + : new Bytes(valueSerializer.serialize(TOPIC, counter))); await("Consumer failed") .atMost(Duration.ofSeconds(30)) @@ -128,9 +123,9 @@ class ApplicationTests .describedAs("Received not all sent events") .isLessThan(100); - assertDoesNotThrow( - () -> endlessConsumer.exitStatus(), - "Consumer should not be running"); + assertThatNoException() + .describedAs("Consumer should not be running") + .isThrownBy(() -> endlessConsumer.exitStatus()); assertThat(endlessConsumer.exitStatus()) .describedAs("Consumer should have exited abnormally") .containsInstanceOf(RecordDeserializationException.class); @@ -286,6 +281,12 @@ class ApplicationTests @Import(ApplicationConfiguration.class) public static class Configuration { + @Bean + Serializer serializer() + { + return new LongSerializer(); + } + @Bean KafkaProducer kafkaProducer(ApplicationProperties properties) {