Palur, First you need to make sure the message is received at Kafka:
message.max.bytes controls the maximum size of a message that can be accepted, and fetch.message.max.bytes controls the maximum number of bytes a consumer issues in one fetch. Guozhang On Mon, Dec 1, 2014 at 7:25 PM, Palur Sandeep <psand...@hawk.iit.edu> wrote: > Thank you so much Jiangle. I got it working. > > I have another problem the consumer doesnt receive message if it is big: > When the producer sends 256kb messages to broker, consumer is able to > retrieve it, but when producer sends 10MB messages to the broker, the > consumer doesn’t receive any message. > > Please tell me how to make the consumer receive 10MB messages. > > On Mon, Dec 1, 2014 at 10:24 AM, Jiangjie Qin <j...@linkedin.com.invalid> > wrote: > > > I think you are printing the class Message instead of MessageAndMetadata. > > The output you got was from Message.toString. > > > > Can you just try something like below? > > > > ... > > ConsumeIterator iter = consumerStream.iterator(); // assuming you have > got > > a consumer stream. > > MessageAndMetadata messageAndMetadta = iter.next(); > > System.out.println(“topic: “ + messageAndMeatadata.topic() + ”partition: > “ > > + messageAndMetadata.partition()); > > > > > > Jiangjie (Becket) Qin > > > > On 11/26/14, 12:56 PM, "Palur Sandeep" <psand...@hawk.iit.edu> wrote: > > > > >Hi Jiangle, > > > > > > > > >Thanks for the information. This is what I get when I print > > >MessageandMetadata > > > > > >*Thread 0: Message(magic = 0, attributes = 0, crc = 127991357, key = > > >java.nio.HeapByteBuffer[pos=0 lim=1 cap=55], payload = > > >java.nio.HeapByteBuffer[pos=0 lim=50 cap=50])* > > > > > >Can you please tell me where can I find partition number in this? > > > > > > > > > > > >On Wed, Nov 26, 2014 at 1:29 PM, Jiangjie Qin <j...@linkedin.com.invalid > > > > >wrote: > > > > > >> Hi Sandeep, > > >> > > >> For old producer, I don’t think you can achieve strict even > distribution > > >> of messages across partitions within the same topic. But You can > > >> potentially reduce the sticking time by setting > > >> topic.metadata.refresh.interval.ms to be lower, e.g. 1 second. > > >> > > >> Kafka-544 added the partition information to MessageAndMetadata. And > > >>that > > >> is back to 11/15/12 so it should have been included in 0.8.1.1. Do you > > >> mean the MessageAndMetadata you got does not partition member or > > >> MessageAndMetadata.partition give you nothing? > > >> > > >> Jiangjie (Becket) Qin > > >> > > >> On 11/26/14, 10:31 AM, "Palur Sandeep" <psand...@hawk.iit.edu> wrote: > > >> > > >> >Hi Jiangjie, > > >> > > > >> >I am using the high level consumer (ZookeeperConsumerConnector), > after > > >> >getting the message from stream, but I don't see this > > >>"message.Partition". > > >> >Please help me how to get the partition id form message. > > >> > > > >> >What is that I can to do get messages evenly distributed among > > >>partitions? > > >> >do you mean that it is not possible in 0.8.1.1 version? > > >> > > > >> >On Wed, Nov 26, 2014 at 12:03 PM, Jiangjie Qin > > >><j...@linkedin.com.invalid > > >> > > > >> >wrote: > > >> > > > >> >> Hi Sandeep, > > >> >> > > >> >> If you are sending messages to different topics, each topic will > > >>stick > > >> >>to > > >> >> a random partition for 10 min. Since they are likely sticking to > > >> >>different > > >> >> brokers, you will still see messages roughly evenly distributed. > > >> >> If you are using high level consumer (ZookeeperConsumerConnector), > > >>after > > >> >> getting the message from stream, you can simply call > > >>message.Partition > > >> >>to > > >> >> get the partition id. > > >> >> > > >> >> Jiangjie (Becket) Qin > > >> >> > > >> >> On 11/25/14, 5:30 PM, "Palur Sandeep" <psand...@hawk.iit.edu> > wrote: > > >> >> > > >> >> >Hi Jiangjie, > > >> >> > > > >> >> >This is what I have understood. Please correct me if I am wrong > > >> >> > > > >> >> >I don¹t use the partition class at all(KeyedMessage<String,String> > > >> >>data = > > >> >> >new KeyedMessage<String, String>(topic_name,new_mes). It > partitions > > >> >> >messages randomly to different partitions. I don¹t see it sticking > > >>to > > >> >>any > > >> >> >broker for 10 mins. I guess it follows some random partitioning > > >>logic. > > >> >>I > > >> >> >am > > >> >> >using the following 0.8.1.1 version. > > >> >> > > > >> >> >MessageAndMetadata on consumer side prints the following message: > > >>Can > > >> >>you > > >> >> >help me find out metadat regarding partition number? > > >> >> > > > >> >> >*Thread 0: Message(magic = 0, attributes = 0, crc = 127991357, > key = > > >> >> >java.nio.HeapByteBuffer[pos=0 lim=1 cap=55], payload = > > >> >> >java.nio.HeapByteBuffer[pos=0 lim=50 cap=50])* > > >> >> > > > >> >> >Thanks > > >> >> >Sandeep > > >> >> > > > >> >> >On Tue, Nov 25, 2014 at 7:07 PM, Jiangjie Qin > > >> >><j...@linkedin.com.invalid> > > >> >> >wrote: > > >> >> > > > >> >> >> Palur, > > >> >> >> > > >> >> >> Just adding to what Guozhang said, the answer to your question > > >>might > > >> >> >> depend on which producer you are using. > > >> >> >> Assuming you are producing messages without keys to the same > > >>topic, > > >> >>in > > >> >> >>new > > >> >> >> producer(KafkaProducer), the messages will go to brokers in a > > >>round > > >> >> >>robin > > >> >> >> way, so the messages will end up in brokers evenly distributed. > > >> >>Whereas > > >> >> >>in > > >> >> >> old producer, it actually sticks to a particular broker for 10 > min > > >> >>(by > > >> >> >> default) then switch to another random partition. In that case, > if > > >> >>you > > >> >> >> send messages fast enough, you might see uneven distribution in > > >> >>brokers. > > >> >> >> > > >> >> >> For the consumer, if you are using high level consumer, when > > >>reading > > >> >> >>from > > >> >> >> KafkaStream, you will get MessageAndMetadata, the topic and > > >>partition > > >> >> >> information is included in it as well as the raw message. > > >> >> >> > > >> >> >> Jiangjie (Becket) Qin > > >> >> >> > > >> >> >> > > >> >> >> > > >> >> >> On 11/25/14, 10:01 AM, "Guozhang Wang" <wangg...@gmail.com> > > wrote: > > >> >> >> > > >> >> >> >Palur, > > >> >> >> > > > >> >> >> >If the 8 partitions are hosted on each one of the nodes, > assuming > > >> >> >> >replication factor 1 then each node will get roughly 100000 / 8 > > >> >> >>messages > > >> >> >> >due to the random partitioner. If you want to know exactly how > > >>many > > >> >> >> >messages is on each broker then you can use a simple consumer > > >>which > > >> >> >>allows > > >> >> >> >you to specify the partition id you want to consume from. > > >> >> >> > > > >> >> >> >In the new consumer (0.9), each of the consumed message will > > >>contain > > >> >> >>the > > >> >> >> >partition id as part of its message metadata. > > >> >> >> > > > >> >> >> >Guozhang > > >> >> >> > > > >> >> >> >On Tue, Nov 25, 2014 at 7:47 AM, Palur Sandeep > > >> >><psand...@hawk.iit.edu> > > >> >> >> >wrote: > > >> >> >> > > > >> >> >> >> Dear Developers, > > >> >> >> >> > > >> >> >> >> I am using the default partitioning logic(Random > Partitioning) > > >>to > > >> >> >> >>produce > > >> >> >> >> messages into brokers. That is I don't use a > partitioner.class. > > >> >> >> >> > > >> >> >> >> My requirement is If I produce 100000 messages using the > below > > >> >>code > > >> >> >>for > > >> >> >> >>a > > >> >> >> >> broker that has 8 partitions across 8 nodes. How many > messages > > >> >>will > > >> >> >>each > > >> >> >> >> partition have? Is there any API that can help me find the > > >>broker > > >> >>id > > >> >> >>of > > >> >> >> >>the > > >> >> >> >> each message I consume from the consumer side? > > >> >> >> >> > > >> >> >> >> PS: I dont want to use partitioner.class. I want use the > > >>kafka's > > >> >> >>default > > >> >> >> >> partitioning logic. > > >> >> >> >> > > >> >> >> >> KeyedMessage<String,String> data = new KeyedMessage<String, > > >> >> >> >> String>(topic_name,new_mes); > > >> >> >> >> > > >> >> >> >> producer.send(data); > > >> >> >> >> > > >> >> >> >> -- > > >> >> >> >> Regards, > > >> >> >> >> Sandeep Palur > > >> >> >> >> Data-Intensive Distributed Systems Laboratory, CS/IIT > > >> >> >> >> Department of Computer Science, Illinois Institute of > > >>Technology > > >> >> >>(IIT) > > >> >> >> >> Phone : 312-647-9833 > > >> >> >> >> Email : psand...@hawk.iit.edu <sraja...@hawk.iit.edu> > > >> >> >> >> > > >> >> >> > > > >> >> >> > > > >> >> >> > > > >> >> >> >-- > > >> >> >> >-- Guozhang > > >> >> >> > > >> >> >> > > >> >> > > > >> >> > > > >> >> >-- > > >> >> >Regards, > > >> >> >Sandeep Palur > > >> >> >Data-Intensive Distributed Systems Laboratory, CS/IIT > > >> >> >Department of Computer Science, Illinois Institute of Technology > > >>(IIT) > > >> >> >Phone : 312-647-9833 > > >> >> >Email : psand...@hawk.iit.edu <sraja...@hawk.iit.edu> > > >> >> > > >> >> > > >> > > > >> > > > >> >-- > > >> >Regards, > > >> >Sandeep Palur > > >> >Data-Intensive Distributed Systems Laboratory, CS/IIT > > >> >Department of Computer Science, Illinois Institute of Technology > (IIT) > > >> >Phone : 312-647-9833 > > >> >Email : psand...@hawk.iit.edu <sraja...@hawk.iit.edu> > > >> > > >> > > > > > > > > >-- > > >Regards, > > >Sandeep Palur > > >Data-Intensive Distributed Systems Laboratory, CS/IIT > > >Department of Computer Science, Illinois Institute of Technology (IIT) > > >Phone : 312-647-9833 > > >Email : psand...@hawk.iit.edu <sraja...@hawk.iit.edu> > > > > > > > -- > Regards, > Sandeep Palur > Data-Intensive Distributed Systems Laboratory, CS/IIT > Department of Computer Science, Illinois Institute of Technology (IIT) > Phone : 312-647-9833 > Email : psand...@hawk.iit.edu <sraja...@hawk.iit.edu> > -- -- Guozhang
