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 static de.juplo.kafka.ApplicationTests.TOPIC;
import static org.assertj.core.api.Assertions.assertThat;
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)
checkSeenOffsetsForProgress();
compareToCommitedOffsets(newOffsets);
});
+
+ assertThrows(
+ IllegalStateException.class,
+ () -> endlessConsumer.exitStatus(),
+ "Consumer should still be running");
}
@Test
assertThat(receivedRecords.size())
.describedAs("Received not all sent events")
.isLessThan(100);
+
+ assertDoesNotThrow(
+ () -> endlessConsumer.exitStatus(),
+ "Consumer should not be running");
+ assertThat(endlessConsumer.exitStatus())
+ .describedAs("Consumer should have exited abnormally")
+ .containsInstanceOf(RecordDeserializationException.class);
}