import org.apache.kafka.common.utils.Bytes;
import org.junit.jupiter.api.*;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
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)
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
@TestPropertySource(
properties = {
- "consumer.bootstrap-server=${spring.embedded.kafka.brokers}",
+ "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}",
"consumer.topic=" + TOPIC })
@EmbeddedKafka(topics = TOPIC, partitions = PARTITIONS)
@Slf4j
@Autowired
KafkaConsumer<Bytes, Bytes> offsetConsumer;
@Autowired
- ApplicationProperties properties;
+ ApplicationProperties applicationProperties;
+ @Autowired
+ KafkaProperties kafkaProperties;
@Autowired
ExecutorService executor;
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
.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);
endlessConsumer =
new EndlessConsumer<>(
executor,
- properties.getClientId(),
- properties.getTopic(),
+ kafkaProperties.getConsumer().getClientId(),
+ applicationProperties.getTopic(),
kafkaConsumer,
captureOffsetAndExecuteTestHandler);
}
@Bean
- KafkaProducer<String, Bytes> kafkaProducer(ApplicationProperties properties)
+ KafkaProducer<String, Bytes> kafkaProducer(KafkaProperties properties)
{
Properties props = new Properties();
- props.put("bootstrap.servers", properties.getBootstrapServer());
+ props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers());
props.put("linger.ms", 100);
props.put("key.serializer", StringSerializer.class.getName());
props.put("value.serializer", BytesSerializer.class.getName());
}
@Bean
- KafkaConsumer<Bytes, Bytes> offsetConsumer(ApplicationProperties properties)
+ KafkaConsumer<Bytes, Bytes> offsetConsumer(KafkaProperties properties)
{
Properties props = new Properties();
- props.put("bootstrap.servers", properties.getBootstrapServer());
+ props.put("bootstrap.servers", properties.getConsumer().getBootstrapServers());
props.put("client.id", "OFFSET-CONSUMER");
- props.put("group.id", properties.getGroupId());
+ props.put("group.id", properties.getConsumer().getGroupId());
props.put("key.deserializer", BytesDeserializer.class.getName());
props.put("value.deserializer", BytesDeserializer.class.getName());