code-not-found/spring-kafka

spring-kafka projects

spring-kafka

This repository contains the source code for the spring-kafka examples posted on https://www.codenotfound.com/spring-kafka/

In case of questions or remarks please leave a comment in the respective blog post. Thanks!

spring-kafka's Issues

Add an example using JUnit5

It would be great, if you could add an example that uses JUnit5 with @EmbeddedKafka.
Thanks! :)

import example.avro.User;

I didnt not find User file in the project
import example.avro.User;

2016-11-28 15:06:01.921 [main] ERROR o.s.boot.SpringApplication - Application startup failed java.lang.NoSuchMethodError: org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.<init>(Lorg/springframework/messaging/converter/MessageConverter;)V

2016-11-28 15:06:01.921 [main] ERROR o.s.boot.SpringApplication - Application startup failed
java.lang.NoSuchMethodError: org.springframework.messaging.handler.annotation.support.MessageMethodArgumentResolver.(Lorg/springframework/messaging/converter/MessageConverter;)V

Getting ClassCast Exception : GenericData$Record cannot be cast to org.apache.avro.specific.SpecificRecordBase

While using the given AvroSerializer and Deserializer in Serde with Kafka streams , I am facing ClassCastException
GenericData$Record cannot be cast to org.apache.avro.specific.SpecificRecordBase.

config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.serdeFrom(new AvroSerializer()
, new AvroDeserializer(MemberMeasureKey.class)).getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.serdeFrom(new AvroSerializer()
, new AvroDeserializer(MemberMeasureValue.class)).getClass());

Please help me on this.

ERROR o.a.zookeeper.server.ZooKeeperServer - ZKShutdownHandler is not registered, so ZooKeeper server won't take any action on ERROR or SHUTDOWN server state changes

log:

. ____ _ __ _ _
/\ / ' __ _ () __ __ _ \ \ \
( ( )__ | '_ | '| | ' / ` | \ \ \
\/ )| |)| | | | | || (| | ) ) ) )
' || .__|| ||| |_, | / / / /
=========||==============|/=////
:: Spring Boot :: (v1.5.4.RELEASE)

16:08:04.256 [main] INFO c.c.kafka.SpringKafkaApplicationTest - Starting SpringKafkaApplicationTest on 55-PC with PID 3796 (started by 55 in D:\Github Code\spring-kafka-master\spring-kafka-multiple-topics)
16:08:04.257 [main] INFO c.c.kafka.SpringKafkaApplicationTest - No active profile set, falling back to default profiles: default
16:08:08.073 [main] INFO c.c.kafka.SpringKafkaApplicationTest - Started SpringKafkaApplicationTest in 4.392 seconds (JVM running for 9.06)
16:08:10.343 [main] INFO c.codenotfound.kafka.producer.Sender - sending payload='Bar [bar=bar]' to topic='spark'
16:08:10.427 [main] INFO c.codenotfound.kafka.producer.Sender - sending payload='Foo [foo=foo]' to topic='stockPrice'
16:08:10.492 [org.springframework.kafka.KafkaListenerEndpointContainer#1-0-C-1] INFO c.c.kafka.consumer.Receiver - received Foo [foo=foo]
16:08:10.492 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO c.c.kafka.consumer.Receiver - received Bar [bar=bar]
16:08:10.492 [main] INFO c.c.kafka.consumer.Receiver - received Bar [bar=null]
16:08:10.492 [main] INFO c.c.kafka.consumer.Receiver - received Foo [foo=null]
16:08:12.107 [main] ERROR o.a.zookeeper.server.ZooKeeperServer - ZKShutdownHandler is not registered, so ZooKeeper server won't take any action on ERROR or SHUTDOWN server state changes
16:08:12.130 [NIOServerCxn.Factory:/127.0.0.1:0] WARN o.a.zookeeper.server.NIOServerCnxn - caught end of stream exception
org.apache.zookeeper.server.ServerCnxn$EndOfStreamException: Unable to read additional data from client sessionid 0x15e65ae64720000, likely client has closed socket
at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:228)
at org.apache.zookeeper.server.NIOServerCnxnFactory.run(NIOServerCnxnFactory.java:203)
at java.lang.Thread.run(Thread.java:748)

Process finished with exit code 0

this is my code:
@test
public void testReceive() throws Exception {
sender.send(BAR_TOPIC, new Bar("bar"));
sender.send(FOO_TOPIC, new Foo("foo"));

receiver.getLatch().await(10000, TimeUnit.MILLISECONDS);
receiver.receiveBar(new Bar());
receiver.receiveFoo(new Foo());

// assertThat(receiver.getLatch().getCount()).isEqualTo(0);
}

I confuse this error.(note: why don't I need to configure zk? why don't I need to configure consumer group.id)

Spring-kafka example without spring-boot

is it possible to use spring-kafka without using spring-boot.
I have implemeneted producer without spring-boot,
but somehow not able to find consumer alternative i.e. @KafkaListener
Any help on this is useful.

java.nio.file.FileSystemException: C:\Users\AppData\Local\Temp\kafka-8364428204801180638\version-2\log.1: The process cannot access the file because it is being used by another process.

I've started the Zookeeper on as shown below

2018-12-16 12:20:58,109 [myid:] - INFO  [main:Environment@100] - Server environment:java.io.tmpdir=C:\Users\e081155\AppData\Local\Temp\
2018-12-16 12:20:58,110 [myid:] - INFO  [main:Environment@100] - Server environment:java.compiler=<NA>
2018-12-16 12:20:58,113 [myid:] - INFO  [main:Environment@100] - Server environment:os.name=Windows 10
2018-12-16 12:20:58,116 [myid:] - INFO  [main:Environment@100] - Server environment:os.arch=amd64
2018-12-16 12:20:58,117 [myid:] - INFO  [main:Environment@100] - Server environment:os.version=10.0
2018-12-16 12:20:58,117 [myid:] - INFO  [main:Environment@100] - Server environment:user.name=e081155
2018-12-16 12:20:58,118 [myid:] - INFO  [main:Environment@100] - Server environment:user.home=C:\Users\e081155
2018-12-16 12:20:58,118 [myid:] - INFO  [main:Environment@100] - Server environment:user.dir=C:\Apache-Kafka\zookeeper-3.4.12\bin
2018-12-16 12:20:58,145 [myid:] - INFO  [main:ZooKeeperServer@835] - tickTime set to 2000
2018-12-16 12:20:58,145 [myid:] - INFO  [main:ZooKeeperServer@844] - minSessionTimeout set to -1
2018-12-16 12:20:58,147 [myid:] - INFO  [main:ZooKeeperServer@853] - maxSessionTimeout set to -1
2018-12-16 12:20:58,382 [myid:] - INFO  [main:ServerCnxnFactory@117] - Using org.apache.zookeeper.server.NIOServerCnxnFactory as server connection factory
2018-12-16 12:20:58,384 [myid:] - INFO  [main:NIOServerCnxnFactory@89] - binding to port 0.0.0.0/0.0.0.0:2181

I'm using Spring Boot version 2.1.1.RELEASE.

2018-12-16 12:21:13.672 ERROR 5976 --- [       Thread-1] org.apache.kafka.test.TestUtils          : Error deleting C:\Users\AppData\Local\Temp\kafka-8364428204801180638

java.nio.file.FileSystemException: C:\Users\AppData\Local\Temp\kafka-8364428204801180638\version-2\log.1: The process cannot access the file because it is being used by another process.

	at sun.nio.fs.WindowsException.translateToIOException(Unknown Source) ~[na:1.8.0_162]
	at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source) ~[na:1.8.0_162]
	at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source) ~[na:1.8.0_162]
	at sun.nio.fs.WindowsFileSystemProvider.implDelete(Unknown Source) ~[na:1.8.0_162]
	at sun.nio.fs.AbstractFileSystemProvider.delete(Unknown Source) ~[na:1.8.0_162]
	at java.nio.file.Files.delete(Unknown Source) ~[na:1.8.0_162]
	at org.apache.kafka.common.utils.Utils$2.visitFile(Utils.java:734) ~[kafka-clients-2.0.1.jar:na]
	at org.apache.kafka.common.utils.Utils$2.visitFile(Utils.java:723) ~[kafka-clients-2.0.1.jar:na]
	at java.nio.file.Files.walkFileTree(Unknown Source) ~[na:1.8.0_162]
	at java.nio.file.Files.walkFileTree(Unknown Source) ~[na:1.8.0_162]
	at org.apache.kafka.common.utils.Utils.delete(Utils.java:723) ~[kafka-clients-2.0.1.jar:na]
	at org.apache.kafka.test.TestUtils$1.run(TestUtils.java:184) ~[kafka-clients-2.0.0-test.jar:na]