My question is very close to Print Kafka Stream Input out to console? with a slight difference: I am using StreamsBuilder instead of Kstreambuilder. (Unfortunately the git with the resolution is not available due to an 404 Error.)
My Listener is:
@StreamListener(LoansStreams.INPUT)
public void handleLoans(@Payload Loans loans) {
final Serde<String> stringSerde = Serdes.String();
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream(stringSerde, stringSerde, "in-stream");
source.flatMapValues(new ValueMapper<String, Iterable<String>>() {
@Override
public Iterable<String> apply(String value) {
ArrayList<String> keywords = new ArrayList<String>();
// apply regex to value and for each match add it to keywords
return keywords;
}
}
}
The error I get is "The method stream(String, Consumed) in the type StreamsBuilder is not applicable for the arguments (Serde, Serde, String)", obviously because it was part of the Kstreambuilder.
Assuming I understood correctly, I must flatmap the received message, but how do I do this? I want to add that it is my first time using Kafka-Cloud-Stream.
My goal is to create a POC with two different microservices access different databases and "connect" both microservices via Kafka in order to apply Saga pattern (or at least start applying its principles).
In case it matter, I started Kafka with only messages to print:
#log4j.rootLogger=WARN, stderr
log4j.rootLogger=OFF, stdout
log4j.appender.stderr=org.apache.log4j.ConsoleAppender
log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
#log4j.appender.stderr.Target=System.err
Learned from Don't print the kafka-console-consumer warnings
Also, based on KeeperErrorCode = NoNode for /admin/preferred_replica_election I am ignoring this zookeeper exception:
[2019-04-30 14:15:11,617] INFO Got user-level KeeperException when processing sessionid:0x100003793980001 type:multi cxid:0x71 zxid:0xbc txntype:-1 reqpath:n/a aborting remaining multi ops. Error Path:/admin/preferred_replica_election Error:KeeperErrorCode = NoNode for /admin/preferred_replica_election (org.apache.zookeeper.server.PrepRequestProcessor)
Last but not least, my sender is:
public void sendLoan(final Loans loans) {
log.info("Sending loans {}", loans);
MessageChannel messageChannel = loansStreams.outboundLoans();
messageChannel.send(MessageBuilder
.withPayload(loans)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
.build());
}
The POM:
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>
<groupId>com.mybank</groupId>
<artifactId>kafka-cloud-stream</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>kafka-cloud-stream</name>
<description>Spring Cloud Stream With Kafka</description>
<properties>
<java.version>11</java.version>
<spring-cloud.version>Greenwich.SR1</spring-cloud.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework/spring-web -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<!-- version>5.1.5.RELEASE</version -->
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-dependencies</artifactId>
<version>${spring-cloud.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
KStreamBuilder is the old builder class that was replaced by StreamsBuilder (in Apache Kafka 1.0.0 release). Not only the name changed, but also method parameters.
As the error message indicates, the new method expects stream(String, Consumed) instead of old stream(Serde, Serde, String).
Compare the corresponding upgrade guide: https://docs.confluent.io/4.0.0/streams/upgrade-guide.html#building-and-running-a-topology