I am getting the below error in kafka like consumer exception
in one of the below error it's printing null what is that null value like bufferUnderFowException.
2022-09-09 20:53:28.644 ERROR 31952 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer : Consumer exception
java.lang.IllegalStateException: This error handler cannot process 'org.apache.kafka.common.protocol.types.SchemaException's; no record information is available
at org.springframework.kafka.listener.DefaultErrorHandler.handleOtherException(DefaultErrorHandler.java:155) ~[spring-kafka-2.8.7.jar:2.8.7]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1799) ~[spring-kafka-2.8.7.jar:2.8.7]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1298) ~[spring-kafka-2.8.7.jar:2.8.7]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.apache.kafka.common.protocol.types.SchemaException: Buffer underflow while parsing response for request with header RequestHeader(apiKey=API_VERSIONS, apiVersion=3, clientId=consumer-group_id-1, correlationId=4)
at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:731) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:874) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:569) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:227) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.awaitMetadataUpdate(ConsumerNetworkClient.java:164) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureCoordinatorReady(AbstractCoordinator.java:258) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.coordinatorUnknownAndUnready(ConsumerCoordinator.java:459) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(ConsumerCoordinator.java:487) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:1262) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1231) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211) ~[kafka-clients-3.1.1.jar:na]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollConsumer(KafkaMessageListenerContainer.java:1522) ~[spring-kafka-2.8.7.jar:2.8.7]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doPoll(KafkaMessageListenerContainer.java:1512) ~[spring-kafka-2.8.7.jar:2.8.7]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:1340) ~[spring-kafka-2.8.7.jar:2.8.7]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1252) ~[spring-kafka-2.8.7.jar:2.8.7]
... 3 common frames omitted
Caused by: java.nio.BufferUnderflowException: null
at java.base/java.nio.Buffer.nextGetIndex(Buffer.java:702) ~[na:na]
at java.base/java.nio.HeapByteBuffer.getInt(HeapByteBuffer.java:433) ~[na:na]
at org.apache.kafka.common.protocol.ByteBufferAccessor.readInt(ByteBufferAccessor.java:43) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.common.message.ResponseHeaderData.read(ResponseHeaderData.java:90) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.common.message.ResponseHeaderData.<init>(ResponseHeaderData.java:66) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.common.requests.ResponseHeader.parse(ResponseHeader.java:71) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.common.requests.AbstractResponse.parseResponse(AbstractResponse.java:100) ~[kafka-clients-3.1.1.jar:na]
at org.apache.kafka.clients.NetworkClient.parseResponse(NetworkClient.java:729) ~[kafka-clients-3.1.1.jar:na]
... 19 common frames omitted
and the code which i wrote is mentioned below i justt created two service classes and in producer service class i am using the tweet model class name so that it will connect to that class.
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class Consumer {
//private final Logger logger = LoggerFactory.getLogger(Producer.class);
@KafkaListener(topics = "Tweet", groupId = "group_id")
public void consume(String message) throws IOException {
System.out.println(String.format("#### -> Consumed message -> %s", message));
}
}
@Service
public class Producer {
// private static final Logger logger = LoggerFactory.getLogger(Producer.class);
private static final String TOPIC = "Tweet";
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String message) {
System.out.println(String.format("#### -> Producing message -> %s", message));
this.kafkaTemplate.send(TOPIC, message);
}
}
and i also autowired producer class in controller class and in console i am also getting the error like There are no in-flight requests for node -1 is these errors are because the kafka port related errors or any firewall related errors or any code errors i am unable to conclude this