Hi Ping,
1) Yes the records are already existent.
2) Yeah the Kafka broker also upgraded to 3.6 version.
That’s why we are upgrading the client code also.
3) Do I need to compare any settings between old broker and new broker?
Please help.
Thanks
Giri
On Mon, 25 Nov 2024 at 6:45 AM, Chia-Ping Tsai <chia7...@gmail.com> wrote:
> hi
>
> 1) The records you want to read are already existent, right?
>
> 2) Are all your code changes solely focused on upgrading the client code
> from version 0.8 to 3.6? Or does the broker get upgraded as well?
>
>
> Thanks,
> Chia-Ping
>
> > giri mungi <girimung...@gmail.com> 於 2024年11月25日 凌晨1:27 寫道：
> >
> > Hi Ping,
> >
> > My bad,commitSync is not required.we can ignore that.
> >
> > Iam calculating diff as below :
> >
> > long stime = Calendar.getInstance().getTimeInMillis();
> > ConsumerRecords <String,String> records =
> > consumer.poll(Duration.ofMillis(2000));
> > long etime = Calendar.getInstance().getTimeInMillis();
> > log.info("Poll Records Count :{} Diff
> :{}",records.count(),(etime-stime));
> >
> > Thanks,
> > Giri
> >
> >> On Sun, Nov 24, 2024 at 10:48 PM Chia-Ping Tsai <chia7...@gmail.com>
> wrote:
> >>
> >> hi Giri
> >>
> >> 1. Why do you call `commitSync`? it seems your application does not use
> the
> >> consumer group as your use case is to random read, right?
> >> 2. how do you calculate "diff"?
> >>
> >> giri mungi <girimung...@gmail.com> 於 2024年11月25日 週一 上午12:50寫道：
> >>
> >>> Hi Ping,
> >>>
> >>> Please find the details below:
> >>>
> >>> 1) Kafka broker version is 3.6.1
> >>>
> >>> 2) Logic Explanation:
> >>>
> >>> Polls messages from Kafka using consumer.poll(Duration.ofMillis(2000)).
> >>>
> >>> *Exit Conditions:* The loop exits when Message Limit > 1000 is reached:
> >>> Then the end flag is true and the loop will exit.
> >>>
> >>>
> >>> boolean end = false;
> >>> consumer.seek(topicPartition, readOffset);
> >>> do {
> >>> JSONObject obj = null;
> >>> long stime = Calendar.getInstance().getTimeInMillis();
> >>> ConsumerRecords < String, String > records =
> >>> consumer.poll(Duration.ofMillis(2000));
> >>> for (ConsumerRecord< String, String > consumerRecord: records) {
> >>> String message = consumerRecord.value();
> >>> obj = new JSONObject(message);
> >>> msglist.add(obj);
> >>> }
> >>> }
> >>> if (msglist.size() >=1000) {
> >>> end = true;
> >>> consumer.commitSync();
> >>> }
> >>> } while (!end);
> >>>
> >>>
> >>> Thanks,
> >>> Giri
> >>>
> >>> On Sun, Nov 24, 2024 at 9:23 PM Chia-Ping Tsai <chia7...@gmail.com>
> >> wrote:
> >>>
> >>>> hi
> >>>>
> >>>> 1) could you share the broker version to us?
> >>>> 2) could you explain how the sample code works? what is the "end"?
> >>>>
> >>>> ```
> >>>> do {
> >>>> ConsumerRecords < String, String > records =
> >>>> consumer.poll(Duration.ofMillis(1500));
> >>>> } while (!end)
> >>>> ```
> >>>>
> >>>> thanks,
> >>>> Chia-Ping
> >>>>
> >>>>
> >>>> giri mungi <girimung...@gmail.com> 於 2024年11月24日 週日 下午11:15寫道：
> >>>>
> >>>>> Hi,
> >>>>>
> >>>>> All of them are from same consumer in each poll.
> >>>>>
> >>>>> Before poll we set offset to user input offset and try to consume
> >> next
> >>>> 1000
> >>>>> messages.
> >>>>>
> >>>>> Thanks
> >>>>> Giridhar.
> >>>>>
> >>>>> On Sun, 24 Nov 2024 at 8:40 PM, Chia-Ping Tsai <chia7...@gmail.com>
> >>>> wrote:
> >>>>>
> >>>>>> hi
> >>>>>>
> >>>>>>> Poll Records Count :0 diff :2004
> >>>>>> Poll Records Count :500 diff :943
> >>>>>>
> >>>>>> Are them from the same consumer in each poll? Or they are based on
> >>>>>> different "offset" and separate consumer's poll?
> >>>>>>
> >>>>>> thanks,
> >>>>>> Chia-Ping
> >>>>>>
> >>>>>>
> >>>>>> giri mungi <girimung...@gmail.com> 於 2024年11月24日 週日 下午8:51寫道：
> >>>>>>
> >>>>>>> Do i need to check any settings in the kafka server level?
> >>>>>>>
> >>>>>>> On Sun, Nov 24, 2024 at 6:19 PM giri mungi <
> >> girimung...@gmail.com>
> >>>>>> wrote:
> >>>>>>>
> >>>>>>>> Hi I have set the below properties as below:
> >>>>>>>>
> >>>>>>>> props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,"1750000");
> >>>>>>>> props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
> >>>>>>>> props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "1500");
> >>>>>>>> props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG,
> >>>>> "1750000");
> >>>>>>>>
> >>>>>>>> Poll Records Count :0 diff :2004
> >>>>>>>> Poll Records Count :500 diff :943
> >>>>>>>> Poll Records Count :0 diff :2000
> >>>>>>>> Poll Records Count :44 diff :3
> >>>>>>>> Poll Records Count :500 diff :205
> >>>>>>>> Poll Records Count :488 diff :1978
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> if i set max records as 500 or 1000 then sometimes i am getting
> >>>> count
> >>>>>> as
> >>>>>>>> "0" and taking max wait ms time which is impacting the overall
> >>>>>> response
> >>>>>>>> time..
> >>>>>>>>
> >>>>>>>> Even though I have mentioned *fetch_min_bytes &
> >> max_poll_records
> >>>> iam
> >>>>>>>> getting "0" records which is impacting the overall response
> >>>> time.*
> >>>>>>>>
> >>>>>>>> How to avoid this please.
> >>>>>>>>
> >>>>>>>> please suggest this.
> >>>>>>>>
> >>>>>>>> Thanks,
> >>>>>>>> Giridar
> >>>>>>>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On Sun, Nov 24, 2024 at 5:26 PM Chia-Ping Tsai <
> >>>> chia7...@apache.org>
> >>>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> hi Giridar
> >>>>>>>>>
> >>>>>>>>> It seems your use case involves random reads, and you expect
> >> the
> >>>>>>> consumer
> >>>>>>>>> to return 1000 records from server at once. Therefore, you
> >> could
> >>>>>>> increase
> >>>>>>>>> the wait time (fetch.max.wait.ms) and fetch size
> >>>> (fetch.min.bytes)
> >>>>> to
> >>>>>>>>> receive a larger response with as many records as possible.
> >>>>>>>>>
> >>>>>>>>> A suitable value for fetch.min.bytes might be 'average size of
> >>>>>> records *
> >>>>>>>>> 1000'. You may also need to increase
> >> max.partition.fetch.bytes.
> >>>> For
> >>>>>>>>> fetch.max.wait.ms, a value of 1500 ms might match your
> >>>>> expectations.
> >>>>>>>>>
> >>>>>>>>> Best,
> >>>>>>>>> Chia-Ping
> >>>>>>>>>
> >>>>>>>>>
> >>>>>>>>> On 2024/11/24 10:55:23 giri mungi wrote:
> >>>>>>>>>> Hi Yang,
> >>>>>>>>>>
> >>>>>>>>>> *This is the old code which is perfectly doing fine and
> >>>> returning
> >>>>>> less
> >>>>>>>>> than
> >>>>>>>>>> 3 seconds for all 1000 records.*
> >>>>>>>>>>
> >>>>>>>>>> do {
> >>>>>>>>>> FetchRequest req = new
> >>>> FetchRequestBuilder().clientId(clientName)
> >>>>>>>>>> .addFetch(a_topic, a_partition, readOffset, fetchSize)
> >>>>>>>>>> .build();
> >>>>>>>>>>
> >>>>>>>>>> FetchResponse fetchResponse = consumer.fetch(req);
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> JSONObject obj = null;
> >>>>>>>>>> ByteBufferMessageSet set = fetchResponse.messageSet(a_topic,
> >>>>>>>>>> a_partition);
> >>>>>>>>>>
> >>>>>>>>>> for (MessageAndOffset messageAndOffset : set) {
> >>>>>>>>>> String message = null;
> >>>>>>>>>> if (messageAndOffset.offset() < readOffset) {
> >>>>>>>>>> log.warn("Found an old offset: {} ,Expecting: {}",
> >>>>>>>>>> messageAndOffset.offset(), readOffset);
> >>>>>>>>>> continue;
> >>>>>>>>>> }
> >>>>>>>>>>
> >>>>>>>>>> ByteBuffer payload = messageAndOffset.message().payload();
> >>>>>>>>>>
> >>>>>>>>>> byte[] bytes = new byte[payload.limit()];
> >>>>>>>>>> payload.get(bytes);
> >>>>>>>>>> try {
> >>>>>>>>>> message = new String(bytes, "UTF-8");
> >>>>>>>>>> } catch (UnsupportedEncodingException ue) {
> >>>>>>>>>> log.warn(ue.getMessage(), ue);
> >>>>>>>>>> message = new String(bytes);
> >>>>>>>>>> }
> >>>>>>>>>>
> >>>>>>>>>> obj = new JSONObject(message);
> >>>>>>>>>> msglist.add(new JSONObject(message));
> >>>>>>>>>> }
> >>>>>>>>>> readOffset = messageAndOffset.nextOffset();
> >>>>>>>>>>
> >>>>>>>>>> if (msglist.size() >= Math.round(limit
> >>>>>>>>>> / inputReq.getApplicationArea().getReqInfo().size())
> >>>>>>>>>> || (endTime - startTime) >= waitTime) {
> >>>>>>>>>> log.info(
> >>>>>>>>>> "Wait condition has been met... exiting the fetch loop.
> >>>>> recordCount
> >>>>>> -
> >>>>>>>>> {},
> >>>>>>>>>> time exhausted - {} ms.",
> >>>>>>>>>> msglist.size(), (endTime - startTime));
> >>>>>>>>>> end = true;
> >>>>>>>>>> }
> >>>>>>>>>>
> >>>>>>>>>> } while (!end);
> >>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>> *The new code is taking 8 seconds.. which is more than
> >> double
> >>>> the
> >>>>>>> time*
> >>>>>>>>>>
> >>>>>>>>>> TopicPartition topicPartition = new TopicPartition("Test",
> >> 0);
> >>>>>>>>>>
> >>>>> consumer.seekToBeginning(Collections.singletonList(topicPartition));
> >>>>>>>>>> long kafkaEarliestOffset =
> >> consumer.position(topicPartition);
> >>>>>>>>>> try (KafkaConsumer < String, String > consumer =
> >>>>>>>>>> KafkaConsumerFactory.createConsumer(clientName, fetchSize))
> >> {
> >>>>>>>>>> consumer.partitionsFor(topicName);
> >>>>>>>>>>
> >>> consumer.assign(Collections.singletonList(topicPartition));
> >>>>>>>>>> consumer.seek(topicPartition, readOffset);
> >>>>>>>>>> do {
> >>>>>>>>>> ConsumerRecords < String, String > records =
> >>>>>>>>>> consumer.poll(Duration.ofMillis(1500));
> >>>>>>>>>> } while (!end)
> >>>>>>>>>>
> >>>>>>>>>> public static KafkaConsumer<String,String>
> >>> createConsumer(String
> >>>>>>>>>> clientName,int fetchSize) {
> >>>>>>>>>> Properties props = new Properties();
> >>>>>>>>>> String kafkaBrokerStr =
> >>>>>>>>>> Config.getConsumerPropValue("kafkabrokerslist");
> >>>>>>>>>> String groupId = Config.getConsumerPropValue("
> >>>>> group.id
> >>>>>> ");
> >>>>>>>>>>
> >>>>>>>>>
> >> props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> >>>>>>>>>> StringDeserializer.class.getName());
> >>>>>>>>>>
> >>>>>>>>>>
> >>>> props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> >>>>>>>>>> StringDeserializer.class.getName());
> >>>>>>>>>>
> >> props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
> >>>> "");
> >>>>>>>>>>
> >>>>>>> props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
> >>>>>>>>>> "false");
> >>>>>>>>>>
> >>>>>> props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> >>>>>>>>>> "earliest");
> >>>>>>>>>> props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,
> >>>>>> "1024");
> >>>>>>>>>>
> >>>>> props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
> >>>>>>>>>>
> >> props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
> >>>>>> "500");
> >>>>>>>>>> return new KafkaConsumer<String,String>(props);
> >>>>>>>>>> }
> >>>>>>>>>>
> >>>>>>>>>> How to improve the performance on this.
> >>>>>>>>>>
> >>>>>>>>>> On Sun, Nov 24, 2024 at 4:11 PM giri mungi <
> >>>> girimung...@gmail.com
> >>>>>>
> >>>>>>>>> wrote:
> >>>>>>>>>>
> >>>>>>>>>>> Hi Yang,
> >>>>>>>>>>>
> >>>>>>>>>>> Can i get the records from kafka as bytes or compression
> >>> form
> >>>> so
> >>>>>>> that
> >>>>>>>>> i
> >>>>>>>>>>> will take less time from kafka.
> >>>>>>>>>>>
> >>>>>>>>>>> I can build messages from those bytes.Is that possible?
> >>>>>>>>>>>
> >>>>>>>>>>> Can you please give suggestions on this.
> >>>>>>>>>>>
> >>>>>>>>>>> Thanks,
> >>>>>>>>>>> Giridar
> >>>>>>>>>>>
> >>>>>>>>>>> On Sun, Nov 24, 2024 at 3:50 PM giri mungi <
> >>>>> girimung...@gmail.com
> >>>>>>>
> >>>>>>>>> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>>> Hi Yang,
> >>>>>>>>>>>>
> >>>>>>>>>>>> Thanks for your reply.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Now what should I do to improve my performance?Because
> >> the
> >>>> old
> >>>>>>> kafka
> >>>>>>>>> code
> >>>>>>>>>>>> was good in performance
> >>>>>>>>>>>>
> >>>>>>>>>>>> These are the properties:
> >>>>>>>>>>>>
> >>>>>>>>>>>> props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "");
> >>>>>>>>>>>>
> >> props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
> >>>>>>>>>>>> "false");
> >>>>>>>>>>>>
> >> props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> >>>>>>>>>>>> "earliest");
> >>>>>>>>>>>> props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,
> >> "1024");
> >>>>>>>>>>>> props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
> >>>>>>>>>>>> props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
> >> "500");
> >>>>>>>>>>>>
> >>>>>>>>>>>> How to decrease this time ,plz suggest .
> >>>>>>>>>>>>
> >>>>>>>>>>>> Thanks,
> >>>>>>>>>>>> Giridar
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Sun, Nov 24, 2024 at 3:15 PM PoAn Yang <
> >>>> yangp...@gmail.com>
> >>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Hi Giridar,
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> *Code explanation:Fetching records is taking time for
> >>> the
> >>>>>> first
> >>>>>>>>> poll.*
> >>>>>>>>>>>>>> Poll Records Count: 500 diff: 1284
> >>>>>>>>>>>>>> Poll Records Count: 500 diff: 3
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> For the first 500 records it took 1284 ms and next 500
> >>>>> records
> >>>>>>> it
> >>>>>>>>> took
> >>>>>>>>>>>>> 4 ms
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *Why this much difference? I would like to improve the
> >>>>>>>>> performance of
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>> first poll time?*
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> IIRC, a consumer has FetchBuffer (cache).
> >>>>>>>>>>>>> If a FetchResponse can return records morn than
> >>>>>>> `max.poll.records`,
> >>>>>>>>> the
> >>>>>>>>>>>>> extra records will be stored in FetchBuffer.
> >>>>>>>>>>>>> That is why the second poll can get data in short time.
> >>>>>>>>>>>>> IMO, we don’t need to balance each poll time, because
> >> that
> >>>>> means
> >>>>>>> we
> >>>>>>>>>>>>> don’t want local cache.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> For example if i give "0" as input offset and it is
> >>> taking
> >>>>>> time
> >>>>>>> as
> >>>>>>>>>>>>> below (6
> >>>>>>>>>>>>>> seconds) and not getting 500 records also,it is
> >> getting
> >>>> only
> >>>>>> 200
> >>>>>>>>>>>>> records
> >>>>>>>>>>>>>> per poll and taking lot of time...why this is
> >> happening
> >>>> and
> >>>>>> how
> >>>>>>> to
> >>>>>>>>>>>>> avoid
> >>>>>>>>>>>>>> this.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken :1227 ms
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken :1181 ms
> >>>>>>>>>>>>>> Poll Records Count :296 Time taken:1234 ms
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken:1133 ms
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *If I give an offset as 110999 and it is getting some
> >>> fast
> >>>>> and
> >>>>>>>>> records
> >>>>>>>>>>>>>> getting as 500 each..Why this difference please.*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Poll Records Count :500 Time taken:1284 ms
> >>>>>>>>>>>>>> Poll Records Count :500 Time taken:3 ms
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> IIUC, a FetchRequest has a limitation from
> >>>> `fetch.max.bytes`.
> >>>>>>>>>>>>> If the record size from offset “0” is bigger than from
> >>>> offset
> >>>>>>>>> “110999”,
> >>>>>>>>>>>>> then a FetchResponse returns less records.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Please correct me if I misunderstood anything.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Thanks,
> >>>>>>>>>>>>> PoAn
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Nov 24, 2024, at 2:09 PM, giri mungi <
> >>>>>> girimung...@gmail.com>
> >>>>>>>>> wrote:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Hi Team,
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Good day to you.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Iam Giridhar.I need your suggestions in kafka
> >>>>>>>>>>>>>> performance improvement please.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *Scenario is: The user will give the offset as input
> >> and
> >>>>> based
> >>>>>>> on
> >>>>>>>>> the
> >>>>>>>>>>>>>> offset we need to give the next 1000 messages from
> >> kafka
> >>>>> topic
> >>>>>>>>> and next
> >>>>>>>>>>>>>> offset.The kafka topic contains only one partition.*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> We are trying to migrate from old kafka to new
> >> kafka.In
> >>>> the
> >>>>>> old
> >>>>>>>>> kafka
> >>>>>>>>>>>>> we
> >>>>>>>>>>>>>> were using code like:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *old code(kafka clients 0.8 .1):*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> FetchRequest req = new
> >>>>>>>>>>>>>>
> >>>> FetchRequestBuilder().clientId(clientName).addFetch(a_topic,
> >>>>>>>>>>>>>> a_partition, readOffset, fetchSize).build();
> >>>>>>>>>>>>>> FetchResponse fetchResponse = consumer.fetch(req);
> >>>>>>>>>>>>>> ByteBufferMessageSet set =
> >>>> fetchResponse.messageSet(a_topic,
> >>>>>>>>>>>>> a_partition);
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> This code is super fast and same we are trying to
> >>> achieve
> >>>>>> using
> >>>>>>>>>>>>>> KafkaConsumer API and getting slowness
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *New kafkaconsumer code is using(kafka clients 3.6
> >> .1)*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> TopicPartition topicPartition = new
> >>> TopicPartition("Test",
> >>>>> 0);
> >>>>>>>>>>>>>>
> >>>>>>>>>
> >>>> consumer.seekToBeginning(Collections.singletonList(topicPartition));
> >>>>>>>>>>>>>> long kafkaEarliestOffset =
> >>>>> consumer.position(topicPartition);
> >>>>>>>>>>>>>> try (KafkaConsumer < String, String > consumer =
> >>>>>>>>>>>>>> KafkaConsumerFactory.createConsumer(clientName,
> >>>> fetchSize))
> >>>>> {
> >>>>>>>>>>>>>>
> >>>>> consumer.assign(Collections.singletonList(topicPartition));
> >>>>>>>>>>>>>> consumer.seek(topicPartition, readOffset);
> >>>>>>>>>>>>>> do {
> >>>>>>>>>>>>>> ConsumerRecords < String, String > records =
> >>>>>>>>>>>>>> consumer.poll(Duration.ofMillis(1500));
> >>>>>>>>>>>>>> } while (!end)
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> public static KafkaConsumer<String,String>
> >>>>>> createConsumer(String
> >>>>>>>>>>>>>> clientName,int fetchSize) {
> >>>>>>>>>>>>>> Properties props = new Properties();
> >>>>>>>>>>>>>> String kafkaBrokerStr =
> >>>>>>>>>>>>>> Config.getConsumerPropValue("kafkabrokerlist");
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>> props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
> >>>>>>>>>>>>>> StringDeserializer.class.getName());
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>> props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
> >>>>>>>>>>>>>> StringDeserializer.class.getName());
> >>>>>>>>>>>>>>
> >>>> props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
> >>>>>>> "");
> >>>>>>>>>>>>>>
> >>>>>>>>> props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
> >>>>>>>>>>>>>> "false");
> >>>>>>>>>>>>>>
> >>>>>>>>> props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
> >>>>>>>>>>>>>> "earliest");
> >>>>>>>>>>>>>>
> >>>> props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,
> >>>>>>>>> "1024");
> >>>>>>>>>>>>>>
> >>>>>>>>> props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"500");
> >>>>>>>>>>>>>>
> >>>>> props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,
> >>>>>>>>> "500");
> >>>>>>>>>>>>>> return new
> >>> KafkaConsumer<String,String>(props);
> >>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *Code explanation:Fetching records is taking time for
> >>> the
> >>>>>> first
> >>>>>>>>> poll.*
> >>>>>>>>>>>>>> Poll Records Count: 500 diff: 1284
> >>>>>>>>>>>>>> Poll Records Count: 500 diff: 3
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> For the first 500 records it took 1284 ms and next 500
> >>>>> records
> >>>>>>> it
> >>>>>>>>> took
> >>>>>>>>>>>>> 4 ms
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *Why this much difference? I would like to improve the
> >>>>>>>>> performance of
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>>> first poll time?*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> 1) How to fetch first 500 records in less time
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *I am also seeing one strange issue.My kafka topic
> >> which
> >>>> has
> >>>>>> one
> >>>>>>>>>>>>> partition
> >>>>>>>>>>>>>> contains some 5 lakh records*.*The starting records
> >> take
> >>>>> more
> >>>>>>>>> time to
> >>>>>>>>>>>>> fetch
> >>>>>>>>>>>>>> from kafka.*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> For example if i give "0" as input offset and it is
> >>> taking
> >>>>>> time
> >>>>>>> as
> >>>>>>>>>>>>> below (6
> >>>>>>>>>>>>>> seconds) and not getting 500 records also,it is
> >> getting
> >>>> only
> >>>>>> 200
> >>>>>>>>>>>>> records
> >>>>>>>>>>>>>> per poll and taking lot of time...why this is
> >> happening
> >>>> and
> >>>>>> how
> >>>>>>> to
> >>>>>>>>>>>>> avoid
> >>>>>>>>>>>>>> this.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken :1227 ms
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken :1181 ms
> >>>>>>>>>>>>>> Poll Records Count :296 Time taken:1234 ms
> >>>>>>>>>>>>>> Poll Records Count :292 Time taken:1133 ms
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> *If I give an offset as 110999 and it is getting some
> >>> fast
> >>>>> and
> >>>>>>>>> records
> >>>>>>>>>>>>>> getting as 500 each..Why this difference please.*
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Poll Records Count :500 Time taken:1284 ms
> >>>>>>>>>>>>>> Poll Records Count :500 Time taken:3 ms
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Please give your suggestion on this.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>> Giridar
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>>
> >>
>
