Spring/spring-boot
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Handle Kafka servers as strings and rely on Kafka's parsing

Browse Source 
Closes gh-34770
This commit is contained in:
Andy Wilkinson
2023-05-02 08:37:23 +01:00
parent dc4efaf276
commit b1d0433d74
7 changed files with 29 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java
View File

@@ -18,7 +18,6 @@ package org.springframework.boot.autoconfigure.kafka;
import java.io.IOException;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.CommonClientConfigs;
@@ -32,7 +31,6 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails.Node;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Jaas;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Retry.Topic;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -190,8 +188,7 @@ public class KafkaAutoConfiguration {
private void applyKafkaConnectionDetailsForConsumer(Map<String, Object> properties,
KafkaConnectionDetails connectionDetails) {
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
nodesToStringList(connectionDetails.getConsumerBootstrapNodes()));
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getConsumerBootstrapServers());
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
}
@@ -199,8 +196,7 @@ public class KafkaAutoConfiguration {
private void applyKafkaConnectionDetailsForProducer(Map<String, Object> properties,
KafkaConnectionDetails connectionDetails) {
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
nodesToStringList(connectionDetails.getProducerBootstrapNodes()));
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getProducerBootstrapServers());
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
}
@@ -208,17 +204,12 @@ public class KafkaAutoConfiguration {
private void applyKafkaConnectionDetailsForAdmin(Map<String, Object> properties,
KafkaConnectionDetails connectionDetails) {
properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
nodesToStringList(connectionDetails.getAdminBootstrapNodes()));
properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getAdminBootstrapNodes());
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
}
}
private List<String> nodesToStringList(List<Node> nodes) {
return nodes.stream().map((node) -> node.host() + ":" + node.port()).toList();
}
private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Topic retryTopic) {
long delay = (retryTopic.getDelay() != null) ? retryTopic.getDelay().toMillis() : 0;
if (delay > 0) {

28
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaConnectionDetails.java
View File

@@ -34,48 +34,38 @@ public interface KafkaConnectionDetails extends ConnectionDetails {
* Returns the list of bootstrap nodes.
* @return the list of bootstrap nodes
*/
List<Node> getBootstrapNodes();
List<String> getBootstrapServers();
/**
* Returns the list of bootstrap nodes used for consumers.
* @return the list of bootstrap nodes used for consumers
*/
default List<Node> getConsumerBootstrapNodes() {
return getBootstrapNodes();
default List<String> getConsumerBootstrapServers() {
return getBootstrapServers();
}
/**
* Returns the list of bootstrap nodes used for producers.
* @return the list of bootstrap nodes used for producers
*/
default List<Node> getProducerBootstrapNodes() {
return getBootstrapNodes();
default List<String> getProducerBootstrapServers() {
return getBootstrapServers();
}
/**
* Returns the list of bootstrap nodes used for the admin.
* @return the list of bootstrap nodes used for the admin
*/
default List<Node> getAdminBootstrapNodes() {
return getBootstrapNodes();
default List<String> getAdminBootstrapNodes() {
return getBootstrapServers();
}
/**
* Returns the list of bootstrap nodes used for Kafka Streams.
* @return the list of bootstrap nodes used for Kafka Streams
*/
default List<Node> getStreamsBootstrapNodes() {
return getBootstrapNodes();
}
/**
* A Kafka node.
*
* @param host the hostname
* @param port the port
*/
record Node(String host, int port) {
default List<String> getStreamsBootstrapServers() {
return getBootstrapServers();
}
}

9
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaStreamsAnnotationDrivenConfiguration.java
View File

@@ -16,7 +16,6 @@
package org.springframework.boot.autoconfigure.kafka;
import java.util.List;
import java.util.Map;
import org.apache.kafka.clients.CommonClientConfigs;
@@ -30,7 +29,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.kafka.KafkaConnectionDetails.Node;
import org.springframework.boot.context.properties.source.InvalidConfigurationPropertyValueException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -87,17 +85,12 @@ class KafkaStreamsAnnotationDrivenConfiguration {
private void applyKafkaConnectionDetailsForStreams(Map<String, Object> properties,
KafkaConnectionDetails connectionDetails) {
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
nodesToStringList(connectionDetails.getStreamsBootstrapNodes()));
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, connectionDetails.getStreamsBootstrapServers());
if (!(connectionDetails instanceof PropertiesKafkaConnectionDetails)) {
properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
}
}
private List<String> nodesToStringList(List<Node> nodes) {
return nodes.stream().map((node) -> node.host() + ":" + node.port()).toList();
}
// Separate class required to avoid BeanCurrentlyInCreationException
static class KafkaStreamsFactoryBeanConfigurer implements InitializingBean {

35
spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/PropertiesKafkaConnectionDetails.java
View File

@@ -27,8 +27,6 @@ import java.util.List;
*/
class PropertiesKafkaConnectionDetails implements KafkaConnectionDetails {
private final int DEFAULT_PORT = 9092;
private final KafkaProperties properties;
PropertiesKafkaConnectionDetails(KafkaProperties properties) {
@@ -36,40 +34,27 @@ class PropertiesKafkaConnectionDetails implements KafkaConnectionDetails {
}
@Override
public List<Node> getBootstrapNodes() {
return asNodes(this.properties.getBootstrapServers());
public List<String> getBootstrapServers() {
return this.properties.getBootstrapServers();
}
@Override
public List<Node> getConsumerBootstrapNodes() {
return bootstrapNodes(this.properties.getConsumer().getBootstrapServers());
public List<String> getConsumerBootstrapServers() {
return getServers(this.properties.getConsumer().getBootstrapServers());
}
@Override
public List<Node> getProducerBootstrapNodes() {
return bootstrapNodes(this.properties.getProducer().getBootstrapServers());
public List<String> getProducerBootstrapServers() {
return getServers(this.properties.getProducer().getBootstrapServers());
}
@Override
public List<Node> getStreamsBootstrapNodes() {
return bootstrapNodes(this.properties.getStreams().getBootstrapServers());
public List<String> getStreamsBootstrapServers() {
return getServers(this.properties.getStreams().getBootstrapServers());
}
private List<Node> bootstrapNodes(List<String> bootstrapServers) {
return (bootstrapServers != null) ? asNodes(bootstrapServers) : getBootstrapNodes();
}
private List<Node> asNodes(List<String> bootstrapServers) {
return bootstrapServers.stream().map(this::asNode).toList();
}
private Node asNode(String bootstrapNode) {
int separatorIndex = bootstrapNode.indexOf(':');
if (separatorIndex == -1) {
return new Node(bootstrapNode, this.DEFAULT_PORT);
}
return new Node(bootstrapNode.substring(0, separatorIndex),
Integer.parseInt(bootstrapNode.substring(separatorIndex + 1)));
private List<String> getServers(List<String> servers) {
return (servers != null) ? servers : getBootstrapServers();
}
}

4
spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java
View File

@@ -839,8 +839,8 @@ class KafkaAutoConfigurationTests {
return new KafkaConnectionDetails() {
@Override
public List<Node> getBootstrapNodes() {
return List.of(new Node("kafka.example.com", 12345));
public List<String> getBootstrapServers() {
return List.of("kafka.example.com:12345");
}
};

6
spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/kafka/KafkaContainerConnectionDetailsFactory.java
View File

@@ -16,7 +16,6 @@
package org.springframework.boot.testcontainers.service.connection.kafka;
import java.net.URI;
import java.util.List;
import org.testcontainers.containers.KafkaContainer;
@@ -53,9 +52,8 @@ class KafkaContainerConnectionDetailsFactory
}
@Override
public List<Node> getBootstrapNodes() {
URI uri = URI.create(getContainer().getBootstrapServers());
return List.of(new Node(uri.getHost(), uri.getPort()));
public List<String> getBootstrapServers() {
return List.of(getContainer().getBootstrapServers());
}
}

6
spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/redpanda/RedpandaContainerConnectionDetailsFactory.java
View File

@@ -16,7 +16,6 @@
package org.springframework.boot.testcontainers.service.connection.redpanda;
import java.net.URI;
import java.util.List;
import org.testcontainers.redpanda.RedpandaContainer;
@@ -52,9 +51,8 @@ class RedpandaContainerConnectionDetailsFactory
}
@Override
public List<Node> getBootstrapNodes() {
URI uri = URI.create(getContainer().getBootstrapServers());
return List.of(new Node(uri.getHost(), uri.getPort()));
public List<String> getBootstrapServers() {
return List.of(getContainer().getBootstrapServers());
}
}