spring-boot - 使用 Spring Cloud Stream Kafka Binder 时无法设置 groupId 和 clientId
问题描述
我在处理 Spring Cloud Stream Kafka Binder 时遇到了严重问题。Spring Cloud 3.0.2.RELEASE 的配置设置存在很多歧义和一致性问题。我一直在尝试为 Kafka 主题设置组 ID 和客户端 ID,但尽管尝试了各种不同的组合,但我无法正确配置组 ID。
该文档声称我们应该能够通过配置以下设置之一来设置组 ID 和客户端 ID: https ://cloud.spring.io/spring-cloud-static/spring-cloud-stream/current/reference /html/spring-cloud-stream.html#binding-properties
spring.cloud.stream.default.group
spring.cloud.stream.default.consumer.group
spring.cloud.stream.kafka.default.consumer.group
spring.cloud.stream.bindings.<channelName>.group
以上配置均不适用于为生产者设置客户端 ID 或为消费者设置组 ID。我得到的唯一进展是通过完全不同的配置设置客户端 ID。
spring.kafka.client-id
spring.kafka.admin.client-id
spring.kafka.producer.client-id
在使用这些设置设置成功设置客户端 ID 后,我尝试为消费者设置组 ID,但令人惊讶的是没有用。
spring.kafka.group-id <---- does not exist as a property, but tried this anyway
spring.kafka.consumer.group-id
编辑:这是应用程序设置。
应用程序.java
@SpringBootApplication
@EnableSwagger2
public class Application {
public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
@Bean
public Docket swaggerApi() {
return new Docket(DocumentationType.SWAGGER_2)
.select()
.apis(RequestHandlerSelectors.any())
.paths(regex("^(?!.*error).+$"))
.build()
.pathMapping("/");
}
}
应用程序.yaml
spring:
cloud:
stream:
bindings:
MyKafkaTopicBinderChannel:
destination: MyKafkaTopic
group: MyServiceGroup
default:
producer:
useNativeEncoding: on
consumer:
useNativeEncoding: on
contentType: application/*+avro
kafka:
binder:
brokers: some.broker.io
producer-properties:
key:
serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
value:
serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
schema:
registry:
url: some.registry.io
consumer-properties:
key:
deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
value:
deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
schema:
registry.url: some.registry.io
specific:
avro:
reader: true
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>io.some.org</groupId>
<artifactId>MyService</artifactId>
<version>1.0.0</version>
<name>chatbotApi</name>
<description>Spring Boot Service</description>
<properties>
<java.version>11</java.version>
<gson.version>2.8.6</gson.version>
<springfox.version>2.9.2</springfox.version>
<swagger-annotations.version>1.6.0</swagger-annotations.version>
<swagger-models.version>1.6.0</swagger-models.version>
<jackson-datatype-jsr310.version>2.10.2</jackson-datatype-jsr310.version>
<avro.version>1.9.2</avro.version>
<avro-maven-plugin.version>1.9.2</avro-maven-plugin.version>
<confluent.kafka.version>5.4.0</confluent.kafka.version>
<kafka-clients.version>2.4.0</kafka-clients.version>
<spring-cloud.version>3.0.2.RELEASE</spring-cloud.version>
</properties>
<repositories>
<repository>
<id>confluent</id>
<url>http://packages.confluent.io/maven/</url>
</repository>
</repositories>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.2.4.RELEASE</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-kafka</artifactId>
<version>${spring-cloud.version}</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.datatype</groupId>
<artifactId>jackson-datatype-jsr310</artifactId>
<version>${jackson-datatype-jsr310.version}</version>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger2</artifactId>
<version>${springfox.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
<exclusion>
<groupId>io.swagger</groupId>
<artifactId>swagger-annotations</artifactId>
</exclusion>
<exclusion>
<groupId>io.swagger</groupId>
<artifactId>swagger-models</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.springfox</groupId>
<artifactId>springfox-swagger-ui</artifactId>
<version>${springfox.version}</version>
</dependency>
<dependency>
<groupId>io.swagger</groupId>
<artifactId>swagger-annotations</artifactId>
<version>${swagger-annotations.version}</version>
</dependency>
<dependency>
<groupId>io.swagger</groupId>
<artifactId>swagger-models</artifactId>
<version>${swagger-models.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-client</artifactId>
<version>${confluent.kafka.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>${confluent.kafka.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-streams-avro-serde</artifactId>
<version>${confluent.kafka.version}</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>${kafka-clients.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>${avro.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro-tools</artifactId>
<version>${avro.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<version>${spring-cloud.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/resources</directory>
<excludes>
<exclude>local.yaml</exclude>
<exclude>avro/*</exclude>
</excludes>
<filtering>true</filtering>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>${avro-maven-plugin.version}</version>
<executions>
<execution>
<id>schemas</id>
<phase>generate-sources</phase>
<goals>
<goal>schema</goal>
<goal>protocol</goal>
<goal>idl-protocol</goal>
</goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/resources/avro/</sourceDirectory>
<outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
</plugins>
</build>
</project>
SpringIntegrationService.java
@Component
@EnableBinding(SpringIntegrationService.KafkaTopicBindings.class)
public class SpringIntegrationService {
private static Logger logger = LoggerFactory.getLogger(SpringIntegrationService.class);
private MessageChannel someChannel;
public interface KafkaTopicBindings {
String MY_KAFKA_TOPIC_BINDER_CHANNEL = "MyKafkaTopicBinderChannel";
@Output(KafkaTopicBindings.MY_KAFKA_TOPIC_BINDER_CHANNEL)
MessageChannel someChannel();
}
public SpringIntegrationService(KafkaTopicBindings bindings) {
this.someChannel = bindings.someChannel();
}
@ServiceActivator(inputChannel = "entry.kafka")
public boolean entryKafka(Message<someChannel> msg) {
logger.info("entryKafka(): Payload: {}", msg.getPayload());
try {
return someChannel.send(MessageBuilder.withPayload(msg.getPayload())
.setHeader(KafkaHeaders.MESSAGE_KEY, "Some Key").build());
} catch (Exception e) {
logger.warn("entryKafka(): Failed to send message onto someChannel topic", e);
return false;
}
}
}
解决方案
推荐阅读
- node.js - 如何知道node js 8.9.1版本支持哪个版本的TLS
- vue.js - Vue.js 中的 $t 是什么
- json - 在 JWS 中使用 JWK
- java - 单击另一个应用程序时 Javafx Stage.setFullScreen() 自动最小化
- c# - C# 反序列化 JSON 列表 WinForm 组合框
- node.js - 监视导入的函数
- google-maps - 如果浏览器位于某些国家/地区的机器上,为什么 Google Maps API 返回 ZERO_RESULTS?
- makefile - 对 `google::protobuf::Message::DebugString[abi:cxx11]() const 的未定义引用
- c# - 在 C# 中检查某个结构的源文件(文本文件)
- symfony4 - FilterService Liip 想象包