I am having serious problems dealing with the Spring Cloud Stream Kafka Binder. There is a lot of ambiguity and consistency issues related the configuration settings for Spring Cloud 3.0.2.RELEASE. I have been trying to set the group ids and the client ids for the Kafka topics, but despite trying out various different combinations, I have not been able to properly configure the group id.
The documentation claims that we should be able to set the group id and the client id by configuring one of the following settings: 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
None of the above configurations work for setting the client id for producers or group id for consumers. The only progress I have gotten at all was setting the client id through a completely different configuration.
spring.kafka.client-id
spring.kafka.admin.client-id
spring.kafka.producer.client-id
After setting success setting client id with those settings, I tried setting the group id for the consumers which surprisingly did not work.
spring.kafka.group-id <---- does not exist as a property, but tried this anyway
spring.kafka.consumer.group-id
Edit: Here is the application setup.
Application.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("/");
}
}
application.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;
}
}
}
Here is a repo where I tried to update the application mentioned in the blog. Cleaned up a few things in the configuration and updated the examples to using the new functional model. I verified that this works. Could you run it on your end and compare with your setup? If you can baseline this sample as a means to report any potential issues, that will help us to further assist you.