SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Initial JUnit 5 migration for Kafka binder

Browse Source
This commit is contained in:
Soby Chacko
2022-03-11 13:38:18 -05:00
parent dedc79d92b
commit 1c4f5e24dc
4 changed files with 31 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,7 +22,7 @@ import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.assertj.core.util.Files;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;

4
binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,7 +25,7 @@ import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.network.SslChannelBuilder;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;

10
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,8 +18,8 @@ package org.springframework.cloud.stream.binder.kafka;
import java.util.Arrays;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@@ -30,7 +30,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaTopicProper
import org.springframework.cloud.stream.config.BindingServiceConfiguration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringExtension;
import static org.assertj.core.api.Assertions.assertThat;
@@ -40,7 +40,7 @@ import static org.assertj.core.api.Assertions.assertThat;
* @since 2.0
*
*/
@RunWith(SpringRunner.class)
@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = { KafkaBinderConfiguration.class,
BindingServiceConfiguration.class })
@TestPropertySource(properties = {

55
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2019 the original author or authors.
* Copyright 2018-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,12 +18,8 @@ package org.springframework.cloud.stream.binder.kafka;
import java.util.Collections;
import kafka.server.KafkaConfig;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.cloud.stream.binder.BinderException;
@@ -34,60 +30,56 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerPro
import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import static org.hamcrest.CoreMatchers.isA;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/**
* @author Soby Chacko
*/
@EmbeddedKafka(brokerProperties = {"auto.create.topics.enable=false"})
public class AutoCreateTopicDisabledTests {
@Rule
public ExpectedException expectedException = ExpectedException.none();
@ClassRule
public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 1)
.brokerProperty(KafkaConfig.AutoCreateTopicsEnableProp(), "false");
private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();
@Test
public void testAutoCreateTopicDisabledFailsOnConsumerIfTopicNonExistentOnBroker()
throws Throwable {
public void testAutoCreateTopicDisabledFailsOnConsumerIfTopicNonExistentOnBroker() {
KafkaProperties kafkaProperties = new TestKafkaProperties();
kafkaProperties.setBootstrapServers(Collections
.singletonList(embeddedKafka.getEmbeddedKafka().getBrokersAsString()));
.singletonList(embeddedKafka.getBrokersAsString()));
KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(
kafkaProperties);
kafkaProperties);
// disable auto create topic on the binder.
configurationProperties.setAutoCreateTopics(false);
KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(
configurationProperties, kafkaProperties, null);
configurationProperties, kafkaProperties, null);
provisioningProvider.setMetadataRetryOperations(new RetryTemplate());
KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(
configurationProperties, provisioningProvider);
configurationProperties, provisioningProvider);
final String testTopicName = "nonExistent" + System.currentTimeMillis();
ExtendedConsumerProperties<KafkaConsumerProperties> properties = new ExtendedConsumerProperties<>(
new KafkaConsumerProperties());
new KafkaConsumerProperties());
expectedException.expect(BinderException.class);
expectedException.expectCause(isA(UnknownTopicOrPartitionException.class));
binder.createConsumerEndpoint(() -> testTopicName, "group", properties);
assertThatExceptionOfType(BinderException.class)
.isThrownBy(() -> binder.createConsumerEndpoint(() -> testTopicName, "group", properties))
.withCauseExactlyInstanceOf(UnknownTopicOrPartitionException.class);
}
@Test
public void testAutoCreateTopicDisabledFailsOnProducerIfTopicNonExistentOnBroker()
throws Throwable {
public void testAutoCreateTopicDisabledFailsOnProducerIfTopicNonExistentOnBroker() {
KafkaProperties kafkaProperties = new TestKafkaProperties();
kafkaProperties.setBootstrapServers(Collections
.singletonList(embeddedKafka.getEmbeddedKafka().getBrokersAsString()));
.singletonList(embeddedKafka.getBrokersAsString()));
KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(
kafkaProperties);
@@ -111,11 +103,8 @@ public class AutoCreateTopicDisabledTests {
ExtendedProducerProperties<KafkaProducerProperties> properties = new ExtendedProducerProperties<>(
new KafkaProducerProperties());
expectedException.expect(BinderException.class);
expectedException.expectCause(isA(UnknownTopicOrPartitionException.class));
binder.bindProducer(testTopicName, new DirectChannel(), properties);
assertThatExceptionOfType(BinderException.class)
.isThrownBy(() -> binder.bindProducer(testTopicName, new DirectChannel(), properties))
.withCauseExactlyInstanceOf(UnknownTopicOrPartitionException.class);
}
}