You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Ratha v <vi...@gmail.com> on 2016/05/03 01:50:09 UTC
Why the consumer not listening to the first message?
Hi all;
In my test program,I start listener. Then sending messages in a loop.
If i send one message, it is not listning that message. If i send 2
messages, it listens one message.If I send 3 , it listens 2 messages..Why
is that?
*Producer*
*KeyedMessage<String, byte[]> message = new KeyedMessage<String,
byte[]>(topic, serializedBytes);*
* if (log.isDebugEnabled()) {*
* log.debug("producing messages to topic : " + topic + "file : " +
payload.get("name"));*
* }*
* for (int i = 0; i < 3; i++) {*
* producer.send(message);*
* System.out.println("producing ..");*
* }*
*Consumer*
*public void run() {*
* try {*
* ConsumerIterator<byte[], byte[]> itr = m_stream.iterator();*
* log.info("Kafka listener is ready to listen..");*
* System.out.println("listens....");*
* while (itr.hasNext()) {*
* byte[] data = itr.next().message();*
*System.out.println("Message received : " + data);*
*}*
*Consumer properties*
enable.auto.commit=true
auto.commit.interval.ms=101
session.timeout.ms=7000
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
zookeeper.connect=zk1.xx\:2181
heartbeat.interval.ms=1000
auto.offset.reset=smallest
serializer.class=kafka.serializer.DefaultEncoder
bootstrap.servers=kk1.xx\:9092
group.id=test
consumer.timeout.ms=-1
fetch.min.bytes=1
receive.buffer.bytes=262144
--
-Ratha
http://vvratha.blogspot.com/
Re: Why the consumer not listening to the first message?
Posted by Ratha v <vi...@gmail.com>.
I fixed this by setting following property in my producer.
request.required.acks=1
On 3 May 2016 at 09:50, Ratha v <vi...@gmail.com> wrote:
> Hi all;
> In my test program,I start listener. Then sending messages in a loop.
> If i send one message, it is not listning that message. If i send 2
> messages, it listens one message.If I send 3 , it listens 2 messages..Why
> is that?
>
> *Producer*
>
> *KeyedMessage<String, byte[]> message = new KeyedMessage<String,
> byte[]>(topic, serializedBytes);*
>
> * if (log.isDebugEnabled()) {*
>
> * log.debug("producing messages to topic : " + topic + "file : " +
> payload.get("name"));*
>
> * }*
>
> * for (int i = 0; i < 3; i++) {*
>
> * producer.send(message);*
>
> * System.out.println("producing ..");*
>
> * }*
> *Consumer*
>
> *public void run() {*
>
> * try {*
>
> * ConsumerIterator<byte[], byte[]> itr = m_stream.iterator();*
>
> * log.info("Kafka listener is ready to listen..");*
>
> * System.out.println("listens....");*
>
> * while (itr.hasNext()) {*
>
> * byte[] data = itr.next().message();*
>
> *System.out.println("Message received : " + data);*
> *}*
>
>
> *Consumer properties*
>
>
> enable.auto.commit=true
>
> auto.commit.interval.ms=101
>
> session.timeout.ms=7000
>
> key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
>
> zookeeper.connect=zk1.xx\:2181
>
> heartbeat.interval.ms=1000
>
> auto.offset.reset=smallest
>
> serializer.class=kafka.serializer.DefaultEncoder
>
> bootstrap.servers=kk1.xx\:9092
>
> group.id=test
>
> consumer.timeout.ms=-1
>
> fetch.min.bytes=1
>
> receive.buffer.bytes=262144
> --
> -Ratha
> http://vvratha.blogspot.com/
>
--
-Ratha
http://vvratha.blogspot.com/