My requirement is to connect a Spring cloud stream application to IBM event streams.
Pom file contains the dependency - spring-cloud-starter-stream-kafka
Controller code
private final GreetingsService greetingsService;
public GreetingsController(GreetingsService greetingsService) {
super();
this.greetingsService = greetingsService;
}
@RequestMapping(value="/sendgreetings",method = RequestMethod.GET)
public void sendGreetings(@RequestParam("message") String message) {
System.out.println("message " + message);
Greetings greetings = new Greetings(System.currentTimeMillis(), message);
greetingsService.sendGreeting(greetings);
}
Greetings Listener class
@Component
public class GreetingsListener
{
private static final Logger LOG = LoggerFactory.getLogger(GreetingsListener.class);
@StreamListener(GreetingsStream.INPUT)
public void handleGreetings(@Payload Greetings greetings) {
LOG.info("Received Greetings {}", greetings);
}
}
Greetings Stream interface
public interface GreetingsStream {
String INPUT = "greetings-in";
String OUTPUT = "greetings-out";
@Input(INPUT)
SubscribableChannel inBoundGreetings();
@Output(OUTPUT)
MessageChannel outBoundGreetings();
}
Greetings Service
@Service
public class GreetingsService {
private final GreetingsStream greetingsStream;
public GreetingsService(GreetingsStream greetingsStream) {
super();
this.greetingsStream = greetingsStream;
}
public void sendGreeting(final Greetings greetings) {
MessageChannel messageChannel = greetingsStream.outBoundGreetings();
messageChannel.send(MessageBuilder.
withPayload(greetings)
//.setHeader(MessageHeaders.CONTENT_TYPE,MimeTypeUtils.APPLICATION_JSON)
.build());
}
}
Greetings Class
@Getter @Setter @ToString @Builder
public class Greetings {
private long timestamp;
private String message;
public Greetings() {
}
public Greetings(long timestamp, String message) {
super();
this.timestamp = timestamp;
this.message = message;
}
}
Binding
@EnableBinding(GreetingsStream.class)
public class EventstreamConfig {}
Main Class
@SpringBootApplication
public class EventstreamApplication {
public static void main(String[] args) {
SpringApplication.run(EventstreamApplication.class, args);
}
}
I am able to connect, send and receive messages to local Kafka instance with the below configurations specified in the properties file
server:
port : 8082
spring:
cloud:
stream:
kafka:
binder:
brokers: localhost:9092
bindings:
greetings-in:
destination: greetings
contentType: application/json
greetings-out:
destination: greetings
contentType: application/json
However i am not able to connect to the IBM event streams on blue mix cloud. Below is my configurations to connect to Event Streams on cloud
spring:
cloud:
stream:
kafka:
binder:
brokers: kafk*****et:9093
consumerProperties:
group-id: foo
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer:org.apache.kafka.common.serialization.StringDeserializer
producerProperties:
client-id: eventstream
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer:org.apache.kafka.common.serialization.StringDeserializer
configuration:
security:
protocol: SASL_SSL
sasl:
mechanism: PLAIN
jaas:
enabled: true
loginModule:org.apache.kafka.common.security.plain.PlainLoginModule
options:
username: ***
password: ***
bindings:
greetings-in:
destination: greetings
contentType: application/json
greetings-out:
destination: greetings
contentType: application/json
Please let me know what is wrong with the configurations. I did not find any example for my requirement.
This is the error i am getting.
2019-12-28 15:57:46.832 WARN 5780 --- [0:0:0:0:1:2181)] org.apache.zookeeper.ClientCnxn : Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_144]
at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source) ~[na:1.8.0_144]
at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) ~[zookeeper-3.4.8.jar:3.4.8--1]
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141) ~[zookeeper-3.4.8.jar:3.4.8--1]
2019-12-28 15:57:46.948 INFO 5780 --- [127.0.0.1:2181)] org.apache.zookeeper.ClientCnxn : Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2019-12-28 15:57:48.973 WARN 5780 --- [127.0.0.1:2181)] org.apache.zookeeper.ClientCnxn : Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_144]
at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source) ~[na:1.8.0_144]
at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) ~[zookeeper-3.4.8.jar:3.4.8--1]
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141) ~[zookeeper-3.4.8.jar:3.4.8--1]
2019-12-28 15:57:50.078 INFO 5780 --- [0:0:0:0:1:2181)] org.apache.zookeeper.ClientCnxn : Opening socket connection to server 0:0:0:0:0:0:0:1/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)
2019-12-28 15:57:52.100 WARN 5780 --- [0:0:0:0:1:2181)] org.apache.zookeeper.ClientCnxn : Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_144]
at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source) ~[na:1.8.0_144]
at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) ~[zookeeper-3.4.8.jar:3.4.8--1]
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141) ~[zookeeper-3.4.8.jar:3.4.8--1]
2019-12-28 15:57:52.201 INFO 5780 --- [127.0.0.1:2181)] org.apache.zookeeper.ClientCnxn : Opening socket connection to server 127.0.0.1/127.0.0.1:2181. Will not attempt to authenticate using SASL (unknown error)
2019-12-28 15:57:54.210 WARN 5780 --- [127.0.0.1:2181)] org.apache.zookeeper.ClientCnxn : Session 0x0 for server null, unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_144]
at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source) ~[na:1.8.0_144]
at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361) ~[zookeeper-3.4.8.jar:3.4.8--1]
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141) ~[zookeeper-3.4.8.jar:3.4.8--1]
2019-12-28 15:57:55.328 INFO 5780 --- [0:0:0:0:1:2181)] org.apache.zookeeper.ClientCnxn : Opening socket connection to server 0:0:0:0:0:0:0:1/0:0:0:0:0:0:0:1:2181. Will not attempt to authenticate using SASL (unknown error)
2019-12-28 15:57:57.458 INFO 5780 --- [ main] org.apache.zookeeper.ZooKeeper : Session: 0x0 closed
2019-12-28 15:57:57.462 WARN 5780 --- [ main] ationConfigEmbeddedWebApplicationContext : Exception encountered during context initialization - cancelling refresh attempt: org.springframework.context.ApplicationContextException: Failed to start bean 'outputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while building outbound endpoint
2019-12-28 15:57:57.467 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Unregistering JMX-exposed beans on shutdown
2019-12-28 15:57:57.467 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Unregistering JMX-exposed beans
2019-12-28 15:57:57.468 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: errorChannel
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: nullChannel
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: greetings-out
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: greetings-in
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: _org.springframework.integration.errorLogger.handler
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.i.monitor.IntegrationMBeanExporter : Summary on shutdown: org.springframework.cloud.stream.binding.StreamListenerMessageHandler@681c0ae6
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans on shutdown
2019-12-28 15:57:57.469 INFO 5780 --- [ main] o.s.j.e.a.AnnotationMBeanExporter : Unregistering JMX-exposed beans
2019-12-28 15:57:57.473 INFO 5780 --- [ main] o.s.s.c.ThreadPoolTaskScheduler : Shutting down ExecutorService 'taskScheduler'
2019-12-28 15:57:57.473 INFO 5780 --- [ main] s.c.a.AnnotationConfigApplicationContext : Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@44bc2449: startup date [Sat Dec 28 15:57:44 IST 2019]; parent: org.springframework.boot.context.embedded.AnnotationConfigEmbeddedWebApplicationContext@67ab1c47
2019-12-28 15:57:57.480 INFO 5780 --- [ main] o.apache.catalina.core.StandardService : Stopping service [Tomcat]
2019-12-28 15:57:57.458 INFO 5780 --- [ain-EventThread] org.apache.zookeeper.ClientCnxn : EventThread shut down for session: 0x0
2019-12-28 15:57:57.500 INFO 5780 --- [ main] utoConfigurationReportLoggingInitializer :
Error starting ApplicationContext. To display the auto-configuration report re-run your application with 'debug' enabled.
2019-12-28 15:57:57.516 ERROR 5780 --- [ main] o.s.boot.SpringApplication : Application startup failed
org.springframework.context.ApplicationContextException: Failed to start bean 'outputBindingLifecycle'; nested exception is org.springframework.cloud.stream.binder.BinderException: Exception thrown while building outbound endpoint
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:176) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:50) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:346) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:149) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:112) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:880) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.finishRefresh(EmbeddedWebApplicationContext.java:144) ~[spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:122) ~[spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:693) [spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:360) [spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:303) [spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1118) [spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1107) [spring-boot-1.5.14.RELEASE.jar:1.5.14.RELEASE]
at com.example.eventstream.EventstreamApplication.main(EventstreamApplication.java:10) [classes/:na]
Caused by: org.springframework.cloud.stream.binder.BinderException: Exception thrown while building outbound endpoint
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:137) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:66) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractBinder.bindProducer(AbstractBinder.java:138) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binding.BindingService.bindProducer(BindingService.java:124) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binding.BindableProxyFactory.bindOutputs(BindableProxyFactory.java:238) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binding.OutputBindingLifecycle.start(OutputBindingLifecycle.java:57) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:173) ~[spring-context-4.3.18.RELEASE.jar:4.3.18.RELEASE]
... 14 common frames omitted
Caused by: org.I0Itec.zkclient.exception.ZkTimeoutException: Unable to connect to zookeeper server within timeout: 10000
at org.I0Itec.zkclient.ZkClient.connect(ZkClient.java:1232) ~[zkclient-0.9.jar:na]
at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:156) ~[zkclient-0.9.jar:na]
at org.I0Itec.zkclient.ZkClient.<init>(ZkClient.java:130) ~[zkclient-0.9.jar:na]
at kafka.utils.ZkUtils$.createZkClientAndConnection(ZkUtils.scala:76) ~[kafka_2.11-0.10.1.1.jar:na]
at kafka.utils.ZkUtils$.apply(ZkUtils.scala:58) ~[kafka_2.11-0.10.1.1.jar:na]
at kafka.utils.ZkUtils.apply(ZkUtils.scala) ~[kafka_2.11-0.10.1.1.jar:na]
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicAndPartitions(KafkaTopicProvisioner.java:171) ~[spring-cloud-stream-binder-kafka-core-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.createTopicsIfAutoCreateEnabledAndAdminUtilsPresent(KafkaTopicProvisioner.java:153) ~[spring-cloud-stream-binder-kafka-core-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:108) ~[spring-cloud-stream-binder-kafka-core-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner.provisionProducerDestination(KafkaTopicProvisioner.java:60) ~[spring-cloud-stream-binder-kafka-core-1.3.2.RELEASE.jar:1.3.2.RELEASE]
at org.springframework.cloud.stream.binder.AbstractMessageChannelBinder.doBindProducer(AbstractMessageChannelBinder.java:119) ~[spring-cloud-stream-1.3.2.RELEASE.jar:1.3.2.RELEASE]
... 20 common frames omitted
