You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by giri mungi <gi...@gmail.com> on 2025/01/16 17:12:43 UTC
Kafka Consumer API with new kafka brokers is taking time.How to avoid plz help.
Hi all,
The below are the two implementation logs : 1) old kafka broker with old
kafka client version (0.8 etc)
2) new kafka brokers (3.6.1) with new kafka client version (3.6.1) .
The old does not have authentication but new kafka we are using
authentication..we went to new one for security and robustness.
*Old kafka brokers with old kafka api implementation: Here we are using *
2025-01-16 08:46:12,380 INFO
(com.cisco.cr.kafka.controller.EventsController:98) [http-nio-8080-exec-8]
- Consume process begin...
2025-01-16 08:46:12,380 INFO
(com.test.cr.kafka.services.NextgenConsumerDriver:82)
[http-nio-8080-exec-8] - Validating request...
2025-01-16 08:46:12,387 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-8]
- Latest offset in Kafka for topic: crt_org_ce_ts1crt & partition: 0 is
16136630 and in user input is 16130077
2025-01-16 08:46:12,387 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:117) [http-nio-8080-exec-8]
- Earliest offset in Kafka for topic: crt_org_ce_ts1crt & partition: 0 is
16129521 and in user input is 16130077
2025-01-16 08:46:12,415 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16130077 - 1043773 KB.
2025-01-16 08:46:12,502 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16130372 - 1043496 KB.
2025-01-16 08:46:12,543 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16130670 - 1044428 KB.
2025-01-16 08:46:12,587 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16130968 - 1044939 KB.
2025-01-16 08:46:12,622 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16131265 - 1045057 KB.
2025-01-16 08:46:12,650 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16131567 - 1042968 KB.
2025-01-16 08:46:12,676 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16131863 - 1043547 KB.
2025-01-16 08:46:12,698 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16132161 - 1045497 KB.
2025-01-16 08:46:12,723 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16132459 - 1043030 KB.
2025-01-16 08:46:12,746 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16132755 - 1045957 KB.
2025-01-16 08:46:12,772 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16133054 - 1043704 KB.
2025-01-16 08:46:12,789 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16133350 - 533178 KB.
2025-01-16 08:46:12,805 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16133499 - 1042816 KB.
2025-01-16 08:46:12,832 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16133797 - 1045895 KB.
2025-01-16 08:46:12,859 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16134093 - 1044584 KB.
2025-01-16 08:46:12,884 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16134398 - 1045775 KB.
2025-01-16 08:46:12,906 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16134702 - 1043018 KB.
2025-01-16 08:46:12,930 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16134999 - 1045228 KB.
2025-01-16 08:46:12,952 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16135290 - 1044921 KB.
2025-01-16 08:46:12,979 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16135579 - 1046197 KB.
2025-01-16 08:46:13,002 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16135876 - 1042978 KB.
2025-01-16 08:46:13,029 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16136172 - 1045985 KB.
2025-01-16 08:46:13,050 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16136463 - 600146 KB.
2025-01-16 08:46:13,057 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16136630 - 0 KB.
2025-01-16 08:46:14,059 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16136630 - 0 KB.
2025-01-16 08:46:15,061 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
- Total size of messages read from offset 16136630 - 0 KB.
2025-01-16 08:46:15,061 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:221) [http-nio-8080-exec-8]
- No messages were found in 3 successive fetches. Stopping the consume
process here.
2025-01-16 08:46:15,061 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:242) [http-nio-8080-exec-8]
- Closing consumer... crt_consumer_crt_org_ce_ts1crt_0_Webex-Webex
2025-01-16 08:46:15,061 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:249) [http-nio-8080-exec-8]
- Summary ::: Total number of messages read from offset 16130077 to
16136630 are - 6553. Count of filtered results - 54
2025-01-16 08:46:15,081 INFO
(com.cisco.cr.kafka.controller.EventsController:154)
[http-nio-8080-exec-8] - Consume process end.
old code: In old code we were using
FetchRequest,FetchResponse,SimpleConsumer without authentication and this
is low level API
public class ChangeEventsKafkaConsumer{
private CEConsumeResponse ceResponse;
public ChangeEventsKafkaConsumer() {
super();
}
public ChangeEventsKafkaConsumer(ConfigBean confBean,
CEConsumeResponse res, long offset) {
super(confBean);
this.ceResponse = res;
this.offset = offset;
}
@Override
public long readMessages(long a_maxRead, String a_topic, int a_partition,
List<String> ng_replicaBrokers, int a_port) throws CustomException {
return 0l;
}
public List<JSONObject> consume(long offset, String a_topic,
int a_partition, List<String> ng_replicaBrokers, int a_port,
CEConsumeRequest inputReq) throws CustomException {
List<JSONObject> msglist = new ArrayList<JSONObject>();
int waitTime = Integer.valueOf(Config
.getConsumerPropValue("pull.wait.time"));
int limit = Integer.valueOf(Config.getConsumerPropValue("pull.size.limit"));
int fetchSize = Integer.valueOf(Config
.getConsumerPropValue("pull.each.fetch.size"));
int emptyLoopLimit = Integer.valueOf(Config
.getConsumerPropValue("pull.empty.loop.limit"));
/*
* Fetching Metadata.
*/
PartitionMetadata metadata = findLeader(ng_replicaBrokers, a_port,
a_topic, a_partition);
if (metadata == null) {
log.error("Can't find metadata for Topic and Partition.");
throw new CustomException(
"Can't find metadata for Topic and Partition.");
}
if (metadata.leader() == null) {
log.error("Can't find Leader for Topic and Partition.");
throw new CustomException(
"Can't find Leader for Topic and Partition.");
}
/*
* Getting lead broker.
*/
String leadBroker = metadata.leader().host();
log.debug("Lead broker for Partion : {}, Topic :{} is {}", a_partition,
a_topic, leadBroker);
SimpleConsumer consumer = new SimpleConsumer(leadBroker, a_port,
configBean.getConnectionTimeOut(),
configBean.getKafkaBufferSize(), clientName);
/*
* Fetching kafka offset from Kafka
*/
long kafkaOffset = getLastOffset(consumer, a_topic, a_partition,
kafka.api.OffsetRequest.LatestTime(), clientName);
/*
* Getting earliest available offset in kafka.
*/
long kafkaEarliestOffset = getLastOffset(consumer, a_topic,
a_partition, kafka.api.OffsetRequest.EarliestTime(), clientName);
/*
* User input offset.
*/
long readOffset = offset;
log.info(
"Latest offset in Kafka for topic: {} & partition: {} is {} and in user
input is {}",
a_topic, a_partition, kafkaOffset, readOffset);
log.info(
"Earliest offset in Kafka for topic: {} & partition: {} is {} and in user
input is {}",
a_topic, a_partition, kafkaEarliestOffset, readOffset);
if (readOffset == 0 && readOffset < kafkaEarliestOffset) {
log.warn("Resetting the offset to earliest available offset in kafka.");
readOffset = kafkaEarliestOffset;
}
boolean end = false;
long startTime = Calendar.getInstance().getTimeInMillis();
long endTime = Calendar.getInstance().getTimeInMillis();
int emptyFetchCount = 0;
do {
FetchRequest req = new FetchRequestBuilder().clientId(clientName)
.addFetch(a_topic, a_partition, readOffset, fetchSize)
.build();
FetchResponse fetchResponse = consumer.fetch(req);
/*
* Reading data from kafka response.
*/
JSONObject obj = null;
ByteBufferMessageSet set = fetchResponse.messageSet(a_topic,
a_partition);
log.info("Total size of messages read from offset {} - {} KB.",
readOffset, set.sizeInBytes());
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);
}
log.debug(
"client name : {} , Offset is : {} , Message is : {} ",
clientName, String.valueOf(messageAndOffset.offset()),
message);
CONSUME_LOG.debug(a_topic + "\t" + a_partition + "\t"
+ String.valueOf(messageAndOffset.offset()));
obj = new JSONObject(message);
if (!inputReq.getApplicationArea().isSubscription()
|| (obj.has("SUBSCRIBER_NAMES") && obj.getString(
"SUBSCRIBER_NAMES").contains(
inputReq.getSender().getReferenceID()))) {
if (inputReq.getApplicationArea().getChangeEventsType() == null
|| (obj.has("NOTIFICATION_EVENT_NAME") && inputReq
.getApplicationArea()
.getChangeEventsType()
.contains(
obj.getString("NOTIFICATION_EVENT_NAME")))) {
msglist.add(new JSONObject(message));
}
}
readOffset = messageAndOffset.nextOffset();
}
endTime = Calendar.getInstance().getTimeInMillis();
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;
} else if (set.sizeInBytes() == 0) {
emptyFetchCount++;
try {
if(emptyFetchCount == emptyLoopLimit) {
log.info("No messages were found in 3 successive fetches. Stopping the
consume process here.");
end = true;
} else {
Thread.sleep(1000);
}
} catch (InterruptedException ie) {
CONSUME_LOG.warn(ie.getMessage(), ie);
}
} else {
emptyFetchCount = 0;
}
req = null;
} while (!end);
*New kafkaconsumer implementation with new kafka brokers:*
2025-01-16 08:46:56,307 INFO
(com.test.cr.kafka.controller.EventsController:99) [http-nio-8080-exec-3]
- Consume process begin...
2025-01-16 08:46:56,307 INFO
(com.test.cr.kafka.services.NextgenConsumerDriver:76)
[http-nio-8080-exec-3] - Validating request...
2025-01-16 08:46:56,307 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:68) [http-nio-8080-exec-3] -
Consume messages Topic: ORG_CHANGE_EVENT_YS1_18_OCT_2024 & partition: 0
2025-01-16 08:46:56,307 INFO
(com.test.cr.kafka.services.KafkaConsumerFactory:45)
[http-nio-8080-exec-3] - KafkaUser:app_client kafkaBrokerStr:
ys1kafka-aln-01.test.com:9092,ys1kafka-aln-02.cisco.com:9092,
ys1kafka-rdn-01.test.com:9092,ys1kafka-rdn-02.test.com:9092
GroupID:crt_org_change_events_lt
2025-01-16 08:46:56,308 INFO
(org.apache.kafka.clients.consumer.ConsumerConfig:370)
[http-nio-8080-exec-3] - ConsumerConfig values:
allow.auto.create.topics = true
auto.commit.interval.ms = 5000
auto.include.jmx.reporter = true
auto.offset.reset = none
bootstrap.servers = [kafka-aln-01.test.com:9092,
kafka-aln-02.test.com:9092, kafka-rdn-01.test.com:9092,
kafka-rdn-02.test.com:9092]
check.crcs = true
client.dns.lookup = use_all_dns_ips
client.id = crt_consumer_http-nio-8080-exec-3
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 10
fetch.min.bytes = 1
group.id =
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
internal.throw.on.fetch.stable.offset.unsupported = false
isolation.level = read_uncommitted
key.deserializer = class
org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class
org.apache.kafka.clients.consumer.RangeAssignor, class
org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = [hidden]
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.connect.timeout.ms = null
sasl.login.read.timeout.ms = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.login.retry.backoff.max.ms = 10000
sasl.login.retry.backoff.ms = 100
sasl.mechanism = PLAIN
sasl.oauthbearer.clock.skew.seconds = 30
sasl.oauthbearer.expected.audience = null
sasl.oauthbearer.expected.issuer = null
sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
sasl.oauthbearer.jwks.endpoint.url = null
sasl.oauthbearer.scope.claim.name = scope
sasl.oauthbearer.sub.claim.name = sub
sasl.oauthbearer.token.endpoint.url = null
security.protocol = SASL_SSL
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 45000
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.3
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class
org.apache.kafka.common.serialization.StringDeserializer
2025-01-16 08:46:56,308 WARN
(org.apache.kafka.clients.consumer.KafkaConsumer:688)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Support for using the
empty group id by consumers is deprecated and will be removed in the next
major release.
2025-01-16 08:46:56,330 INFO
(org.apache.kafka.common.utils.AppInfoParser:119) [http-nio-8080-exec-3] -
Kafka version: 3.6.1
2025-01-16 08:46:56,330 INFO
(org.apache.kafka.common.utils.AppInfoParser:120) [http-nio-8080-exec-3] -
Kafka commitId: 5e3c2b738d253ff5
2025-01-16 08:46:56,330 INFO
(org.apache.kafka.common.utils.AppInfoParser:121) [http-nio-8080-exec-3] -
Kafka startTimeMs: 1737046016330
2025-01-16 08:46:56,330 INFO
(org.apache.kafka.clients.consumer.KafkaConsumer:1067)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Assigned to
partition(s): ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
2025-01-16 08:46:56,330 INFO
(org.apache.kafka.clients.consumer.KafkaConsumer:1067)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Assigned to
partition(s): ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
2025-01-16 08:46:56,832 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:95) [http-nio-8080-exec-3] -
EarliestOffset :0 LatestOffset :16136566 Time :502 ms
2025-01-16 08:46:56,832 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:97) [http-nio-8080-exec-3] -
Latest & Earliest offset in Kafka for topic:
ORG_CHANGE_EVENT_YS1_18_OCT_2024 & partition: 0 Latest is 16136566 Earliest
is 0 and in user input is 16130077
2025-01-16 08:46:56,832 INFO
(org.apache.kafka.clients.consumer.KafkaConsumer:1564)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Seeking to offset
16130077 for partition ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
2025-01-16 08:46:57,157 INFO (org.apache.kafka.clients.Metadata:287)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Cluster ID:
WyVVmQEwTbeRp73DYJgk7Q
2025-01-16 08:46:58,741 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:46:58,823 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:46:58,898 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:46:58,964 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :188
2025-01-16 08:47:00,243 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:00,316 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:00,383 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:00,455 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :235
2025-01-16 08:47:01,735 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:01,799 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:01,966 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:02,035 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :294
2025-01-16 08:47:02,912 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:02,981 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :500
2025-01-16 08:47:03,050 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :272
2025-01-16 08:47:05,087 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :0
2025-01-16 08:47:08,088 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :0
2025-01-16 08:47:11,088 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
- Poll Records Count :0
2025-01-16 08:47:11,088 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:157) [http-nio-8080-exec-3]
- No messages were found in 3 successive fetches. Stopping the consume
process here.
2025-01-16 08:47:11,088 INFO
(com.test.cr.kafka.services.TestKafkaConsumer:179) [http-nio-8080-exec-3]
- Summary ::: Total number of messages read from offset 16130077 to
16136566 are - 6489. Count of filtered results - 54
2025-01-16 08:47:11,088 INFO
(org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:1025)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Resetting generation
and member id due to: consumer pro-actively leaving the group
2025-01-16 08:47:11,089 INFO
(org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:1072)
[http-nio-8080-exec-3] - [Consumer
clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Request joining group
due to: consumer pro-actively leaving the group
2025-01-16 08:47:11,138 INFO (org.apache.kafka.common.metrics.Metrics:693)
[http-nio-8080-exec-3] - Metrics scheduler closed
2025-01-16 08:47:11,139 INFO (org.apache.kafka.common.metrics.Metrics:697)
[http-nio-8080-exec-3] - Closing reporter
org.apache.kafka.common.metrics.JmxReporter
2025-01-16 08:47:11,139 INFO (org.apache.kafka.common.metrics.Metrics:703)
[http-nio-8080-exec-3] - Metrics reporters closed
2025-01-16 08:47:11,140 INFO
(org.apache.kafka.common.utils.AppInfoParser:83) [http-nio-8080-exec-3] -
App info kafka.consumer for crt_consumer_http-nio-8080-exec-3 unregistered
2025-01-16 08:47:11,159 INFO
(com.cisco.cr.kafka.controller.EventsController:156)
[http-nio-8080-exec-3] - Consume process end.
New code:Here we are using KafkaConsumer API with SASL authentication
public List<JSONObject> consume(long offset, String topicName,int
partition,CEConsumeRequest inputReq) throws CustomException {
List<JSONObject> msglist = new ArrayList<JSONObject>();
int waitTime = Integer.valueOf(Config
.getConsumerPropValue("pull.wait.time"));
int limit = Integer.valueOf(Config.getConsumerPropValue("pull.size.limit"));
int emptyLoopLimit = Integer.valueOf(Config
.getConsumerPropValue("pull.empty.loop.limit"));
int fetchSize = Integer.valueOf(Config
.getConsumerPropValue("pull.each.fetch.size"));
log.info(
"Consume messages Topic: {} & partition: {}",
topicName, partition);
TopicPartition topicPartition = new TopicPartition(topicName, partition);
long readOffset = offset;
clientName = "crt_consumer_" + Thread.currentThread().getName();
try (KafkaConsumer<String,String> consumer =
KafkaConsumerFactory.createConsumer(clientName,fetchSize,topicPartition)){
consumer.assign(Collections.singletonList(topicPartition));
long OffsetStartTime = System.currentTimeMillis();
long OffsetEndTime= System.currentTimeMillis();
long kafkaEarliestOffset=0;
long latestOffset=0;
Admin adminClient = KafkaConsumerFactory.getAdminClient();
try {
kafkaEarliestOffset = adminClient.listOffsets(
Map.of(topicPartition, OffsetSpec.earliest())
).all().get().get(topicPartition).offset();
latestOffset = adminClient.listOffsets(
Map.of(topicPartition, OffsetSpec.latest())
).all().get().get(topicPartition).offset();
OffsetEndTime = System.currentTimeMillis();
} catch (Exception e) {
e.printStackTrace();
}
log.info("EarliestOffset :{} LatestOffset :{} Time :{}
ms",kafkaEarliestOffset,latestOffset,(OffsetEndTime - OffsetStartTime));
log.info(
"Latest & Earliest offset in Kafka for topic: {} & partition: {} Latest
is {} Earliest is {} and in user input is {}",
topicName, partition, latestOffset, kafkaEarliestOffset,readOffset);
if (readOffset == 0 && readOffset < kafkaEarliestOffset) {
log.warn("Resetting the offset to earliest available offset in kafka.");
readOffset = kafkaEarliestOffset;
}
boolean end = false;
long startTime = System.currentTimeMillis();
long endTime = System.currentTimeMillis();
int emptyFetchCount = 0;
consumer.seek(topicPartition, offset);
do {
JSONObject obj = null;
ConsumerRecords < String, String > records =
consumer.poll(Duration.ofMillis(2000));
log.info("Poll Records Count :{}",records.count());
for (ConsumerRecord< String, String > consumerRecord: records) {
long currentOffset = consumerRecord.offset();
if (currentOffset < readOffset) {
log.warn("Found an old offset: {}, Expecting: {}",
currentOffset, readOffset);
continue;
}
String message = consumerRecord.value();
log.debug(
"client name : {} , Offset is : {} , Message is : {} ",
clientName, readOffset,
message);
CONSUME_LOG.debug(topicName + "\t" + partition + "\t"
+ String.valueOf(currentOffset));
obj = new JSONObject(message);
if (!inputReq.getApplicationArea().isSubscription()
|| (obj.has("SUBSCRIBER_NAMES") && obj.getString(
"SUBSCRIBER_NAMES").contains(
inputReq.getSender().getReferenceID()))) {
if (inputReq.getApplicationArea().getChangeEventsType() == null
|| (obj.has("NOTIFICATION_EVENT_NAME") && inputReq
.getApplicationArea()
.getChangeEventsType()
.contains(
obj.getString("NOTIFICATION_EVENT_NAME")))) {
msglist.add(obj);
}
}
readOffset = currentOffset + 1;
}
endTime = System.currentTimeMillis();
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;
//consumer.commitSync();
} else if (records.isEmpty()) {
emptyFetchCount++;
try {
if(emptyFetchCount == emptyLoopLimit) {
log.info("No messages were found in 3 successive fetches. Stopping the
consume process here.");
end = true;
} else {
Thread.sleep(1000);
}
} catch (InterruptedException ie) {
CONSUME_LOG.warn(ie.getMessage(), ie);
}
} else {
emptyFetchCount = 0;
}
} while (!end);
The new kafka consumer API is very slow..i have tried lot of settings like
increasing MAX_POLL_RECORDS_CONFIG,FETCH_MIN_BYTES_CONFIG etc..but no use.
Is this time expected with the new KafkaConsumer API.How to improve my
timings.
please help on this.
Regards,
Giri
Re: Kafka Consumer API with new kafka brokers is taking time.How to avoid plz help.
Posted by giri mungi <gi...@gmail.com>.
The old implementation is taking upto 2.5 seconds.
new implementation is taking 10 seconds (10 polls for each poll have 500
records returning)
Authentication happens every time internally?is this delay expected? How to
explain this to management please?
On Thu, Jan 16, 2025 at 10:42 PM giri mungi <gi...@gmail.com> wrote:
> Hi all,
>
>
> The below are the two implementation logs : 1) old kafka broker with old
> kafka client version (0.8 etc)
> 2) new kafka brokers (3.6.1) with new kafka client version (3.6.1) .
>
> The old does not have authentication but new kafka we are using
> authentication..we went to new one for security and robustness.
>
>
>
> *Old kafka brokers with old kafka api implementation: Here we are using *
>
> 2025-01-16 08:46:12,380 INFO
> (com.cisco.cr.kafka.controller.EventsController:98) [http-nio-8080-exec-8]
> - Consume process begin...
> 2025-01-16 08:46:12,380 INFO
> (com.test.cr.kafka.services.NextgenConsumerDriver:82)
> [http-nio-8080-exec-8] - Validating request...
> 2025-01-16 08:46:12,387 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-8]
> - Latest offset in Kafka for topic: crt_org_ce_ts1crt & partition: 0 is
> 16136630 and in user input is 16130077
> 2025-01-16 08:46:12,387 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:117) [http-nio-8080-exec-8]
> - Earliest offset in Kafka for topic: crt_org_ce_ts1crt & partition: 0 is
> 16129521 and in user input is 16130077
> 2025-01-16 08:46:12,415 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16130077 - 1043773 KB.
> 2025-01-16 08:46:12,502 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16130372 - 1043496 KB.
> 2025-01-16 08:46:12,543 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16130670 - 1044428 KB.
> 2025-01-16 08:46:12,587 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16130968 - 1044939 KB.
> 2025-01-16 08:46:12,622 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16131265 - 1045057 KB.
> 2025-01-16 08:46:12,650 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16131567 - 1042968 KB.
> 2025-01-16 08:46:12,676 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16131863 - 1043547 KB.
> 2025-01-16 08:46:12,698 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16132161 - 1045497 KB.
> 2025-01-16 08:46:12,723 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16132459 - 1043030 KB.
> 2025-01-16 08:46:12,746 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16132755 - 1045957 KB.
> 2025-01-16 08:46:12,772 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16133054 - 1043704 KB.
> 2025-01-16 08:46:12,789 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16133350 - 533178 KB.
> 2025-01-16 08:46:12,805 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16133499 - 1042816 KB.
> 2025-01-16 08:46:12,832 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16133797 - 1045895 KB.
> 2025-01-16 08:46:12,859 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16134093 - 1044584 KB.
> 2025-01-16 08:46:12,884 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16134398 - 1045775 KB.
> 2025-01-16 08:46:12,906 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16134702 - 1043018 KB.
> 2025-01-16 08:46:12,930 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16134999 - 1045228 KB.
> 2025-01-16 08:46:12,952 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16135290 - 1044921 KB.
> 2025-01-16 08:46:12,979 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16135579 - 1046197 KB.
> 2025-01-16 08:46:13,002 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16135876 - 1042978 KB.
> 2025-01-16 08:46:13,029 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16136172 - 1045985 KB.
> 2025-01-16 08:46:13,050 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16136463 - 600146 KB.
> 2025-01-16 08:46:13,057 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16136630 - 0 KB.
> 2025-01-16 08:46:14,059 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16136630 - 0 KB.
> 2025-01-16 08:46:15,061 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:167) [http-nio-8080-exec-8]
> - Total size of messages read from offset 16136630 - 0 KB.
> 2025-01-16 08:46:15,061 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:221) [http-nio-8080-exec-8]
> - No messages were found in 3 successive fetches. Stopping the consume
> process here.
> 2025-01-16 08:46:15,061 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:242) [http-nio-8080-exec-8]
> - Closing consumer... crt_consumer_crt_org_ce_ts1crt_0_Webex-Webex
> 2025-01-16 08:46:15,061 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:249) [http-nio-8080-exec-8]
> - Summary ::: Total number of messages read from offset 16130077 to
> 16136630 are - 6553. Count of filtered results - 54
> 2025-01-16 08:46:15,081 INFO
> (com.cisco.cr.kafka.controller.EventsController:154)
> [http-nio-8080-exec-8] - Consume process end.
>
> old code: In old code we were using
> FetchRequest,FetchResponse,SimpleConsumer without authentication and this
> is low level API
>
>
> public class ChangeEventsKafkaConsumer{
>
>
> private CEConsumeResponse ceResponse;
>
> public ChangeEventsKafkaConsumer() {
> super();
> }
>
> public ChangeEventsKafkaConsumer(ConfigBean confBean,
> CEConsumeResponse res, long offset) {
> super(confBean);
> this.ceResponse = res;
> this.offset = offset;
> }
>
> @Override
> public long readMessages(long a_maxRead, String a_topic, int a_partition,
> List<String> ng_replicaBrokers, int a_port) throws CustomException {
> return 0l;
> }
>
> public List<JSONObject> consume(long offset, String a_topic,
> int a_partition, List<String> ng_replicaBrokers, int a_port,
> CEConsumeRequest inputReq) throws CustomException {
> List<JSONObject> msglist = new ArrayList<JSONObject>();
>
> int waitTime = Integer.valueOf(Config
> .getConsumerPropValue("pull.wait.time"));
> int limit =
> Integer.valueOf(Config.getConsumerPropValue("pull.size.limit"));
> int fetchSize = Integer.valueOf(Config
> .getConsumerPropValue("pull.each.fetch.size"));
> int emptyLoopLimit = Integer.valueOf(Config
> .getConsumerPropValue("pull.empty.loop.limit"));
>
> /*
> * Fetching Metadata.
> */
> PartitionMetadata metadata = findLeader(ng_replicaBrokers, a_port,
> a_topic, a_partition);
> if (metadata == null) {
> log.error("Can't find metadata for Topic and Partition.");
> throw new CustomException(
> "Can't find metadata for Topic and Partition.");
> }
> if (metadata.leader() == null) {
> log.error("Can't find Leader for Topic and Partition.");
> throw new CustomException(
> "Can't find Leader for Topic and Partition.");
> }
>
> /*
> * Getting lead broker.
> */
> String leadBroker = metadata.leader().host();
> log.debug("Lead broker for Partion : {}, Topic :{} is {}", a_partition,
> a_topic, leadBroker);
> SimpleConsumer consumer = new SimpleConsumer(leadBroker, a_port,
> configBean.getConnectionTimeOut(),
> configBean.getKafkaBufferSize(), clientName);
>
> /*
> * Fetching kafka offset from Kafka
> */
> long kafkaOffset = getLastOffset(consumer, a_topic, a_partition,
> kafka.api.OffsetRequest.LatestTime(), clientName);
>
> /*
> * Getting earliest available offset in kafka.
> */
> long kafkaEarliestOffset = getLastOffset(consumer, a_topic,
> a_partition, kafka.api.OffsetRequest.EarliestTime(), clientName);
>
> /*
> * User input offset.
> */
> long readOffset = offset;
>
> log.info(
> "Latest offset in Kafka for topic: {} & partition: {} is {} and in user
> input is {}",
> a_topic, a_partition, kafkaOffset, readOffset);
> log.info(
> "Earliest offset in Kafka for topic: {} & partition: {} is {} and in user
> input is {}",
> a_topic, a_partition, kafkaEarliestOffset, readOffset);
>
> if (readOffset == 0 && readOffset < kafkaEarliestOffset) {
> log.warn("Resetting the offset to earliest available offset in kafka.");
> readOffset = kafkaEarliestOffset;
> }
>
> boolean end = false;
> long startTime = Calendar.getInstance().getTimeInMillis();
> long endTime = Calendar.getInstance().getTimeInMillis();
> int emptyFetchCount = 0;
> do {
> FetchRequest req = new FetchRequestBuilder().clientId(clientName)
> .addFetch(a_topic, a_partition, readOffset, fetchSize)
> .build();
>
> FetchResponse fetchResponse = consumer.fetch(req);
>
>
>
> /*
> * Reading data from kafka response.
> */
> JSONObject obj = null;
> ByteBufferMessageSet set = fetchResponse.messageSet(a_topic,
> a_partition);
> log.info("Total size of messages read from offset {} - {} KB.",
> readOffset, set.sizeInBytes());
> 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);
> }
> log.debug(
> "client name : {} , Offset is : {} , Message is : {} ",
> clientName, String.valueOf(messageAndOffset.offset()),
> message);
> CONSUME_LOG.debug(a_topic + "\t" + a_partition + "\t"
> + String.valueOf(messageAndOffset.offset()));
> obj = new JSONObject(message);
> if (!inputReq.getApplicationArea().isSubscription()
> || (obj.has("SUBSCRIBER_NAMES") && obj.getString(
> "SUBSCRIBER_NAMES").contains(
> inputReq.getSender().getReferenceID()))) {
> if (inputReq.getApplicationArea().getChangeEventsType() == null
> || (obj.has("NOTIFICATION_EVENT_NAME") && inputReq
> .getApplicationArea()
> .getChangeEventsType()
> .contains(
> obj.getString("NOTIFICATION_EVENT_NAME")))) {
> msglist.add(new JSONObject(message));
> }
> }
> readOffset = messageAndOffset.nextOffset();
> }
> endTime = Calendar.getInstance().getTimeInMillis();
> 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;
> } else if (set.sizeInBytes() == 0) {
> emptyFetchCount++;
> try {
> if(emptyFetchCount == emptyLoopLimit) {
> log.info("No messages were found in 3 successive fetches. Stopping the
> consume process here.");
> end = true;
> } else {
> Thread.sleep(1000);
> }
> } catch (InterruptedException ie) {
> CONSUME_LOG.warn(ie.getMessage(), ie);
> }
> } else {
> emptyFetchCount = 0;
> }
> req = null;
> } while (!end);
>
> *New kafkaconsumer implementation with new kafka brokers:*
>
>
> 2025-01-16 08:46:56,307 INFO
> (com.test.cr.kafka.controller.EventsController:99) [http-nio-8080-exec-3]
> - Consume process begin...
> 2025-01-16 08:46:56,307 INFO
> (com.test.cr.kafka.services.NextgenConsumerDriver:76)
> [http-nio-8080-exec-3] - Validating request...
> 2025-01-16 08:46:56,307 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:68) [http-nio-8080-exec-3] -
> Consume messages Topic: ORG_CHANGE_EVENT_YS1_18_OCT_2024 & partition: 0
> 2025-01-16 08:46:56,307 INFO
> (com.test.cr.kafka.services.KafkaConsumerFactory:45)
> [http-nio-8080-exec-3] - KafkaUser:app_client kafkaBrokerStr:
> ys1kafka-aln-01.test.com:9092,ys1kafka-aln-02.cisco.com:9092,
> ys1kafka-rdn-01.test.com:9092,ys1kafka-rdn-02.test.com:9092
> GroupID:crt_org_change_events_lt
> 2025-01-16 08:46:56,308 INFO
> (org.apache.kafka.clients.consumer.ConsumerConfig:370)
> [http-nio-8080-exec-3] - ConsumerConfig values:
> allow.auto.create.topics = true
> auto.commit.interval.ms = 5000
> auto.include.jmx.reporter = true
> auto.offset.reset = none
> bootstrap.servers = [kafka-aln-01.test.com:9092,
> kafka-aln-02.test.com:9092, kafka-rdn-01.test.com:9092,
> kafka-rdn-02.test.com:9092]
> check.crcs = true
> client.dns.lookup = use_all_dns_ips
> client.id = crt_consumer_http-nio-8080-exec-3
> client.rack =
> connections.max.idle.ms = 540000
> default.api.timeout.ms = 60000
> enable.auto.commit = false
> exclude.internal.topics = true
> fetch.max.bytes = 52428800
> fetch.max.wait.ms = 10
> fetch.min.bytes = 1
> group.id =
> group.instance.id = null
> heartbeat.interval.ms = 3000
> interceptor.classes = []
> internal.leave.group.on.close = true
> internal.throw.on.fetch.stable.offset.unsupported = false
> isolation.level = read_uncommitted
> key.deserializer = class
> org.apache.kafka.common.serialization.StringDeserializer
> max.partition.fetch.bytes = 1048576
> max.poll.interval.ms = 300000
> max.poll.records = 500
> metadata.max.age.ms = 300000
> metric.reporters = []
> metrics.num.samples = 2
> metrics.recording.level = INFO
> metrics.sample.window.ms = 30000
> partition.assignment.strategy = [class
> org.apache.kafka.clients.consumer.RangeAssignor, class
> org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
> receive.buffer.bytes = 65536
> reconnect.backoff.max.ms = 1000
> reconnect.backoff.ms = 50
> request.timeout.ms = 30000
> retry.backoff.ms = 100
> sasl.client.callback.handler.class = null
> sasl.jaas.config = [hidden]
> sasl.kerberos.kinit.cmd = /usr/bin/kinit
> sasl.kerberos.min.time.before.relogin = 60000
> sasl.kerberos.service.name = null
> sasl.kerberos.ticket.renew.jitter = 0.05
> sasl.kerberos.ticket.renew.window.factor = 0.8
> sasl.login.callback.handler.class = null
> sasl.login.class = null
> sasl.login.connect.timeout.ms = null
> sasl.login.read.timeout.ms = null
> sasl.login.refresh.buffer.seconds = 300
> sasl.login.refresh.min.period.seconds = 60
> sasl.login.refresh.window.factor = 0.8
> sasl.login.refresh.window.jitter = 0.05
> sasl.login.retry.backoff.max.ms = 10000
> sasl.login.retry.backoff.ms = 100
> sasl.mechanism = PLAIN
> sasl.oauthbearer.clock.skew.seconds = 30
> sasl.oauthbearer.expected.audience = null
> sasl.oauthbearer.expected.issuer = null
> sasl.oauthbearer.jwks.endpoint.refresh.ms = 3600000
> sasl.oauthbearer.jwks.endpoint.retry.backoff.max.ms = 10000
> sasl.oauthbearer.jwks.endpoint.retry.backoff.ms = 100
> sasl.oauthbearer.jwks.endpoint.url = null
> sasl.oauthbearer.scope.claim.name = scope
> sasl.oauthbearer.sub.claim.name = sub
> sasl.oauthbearer.token.endpoint.url = null
> security.protocol = SASL_SSL
> security.providers = null
> send.buffer.bytes = 131072
> session.timeout.ms = 45000
> socket.connection.setup.timeout.max.ms = 30000
> socket.connection.setup.timeout.ms = 10000
> ssl.cipher.suites = null
> ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
> ssl.endpoint.identification.algorithm = https
> ssl.engine.factory.class = null
> ssl.key.password = null
> ssl.keymanager.algorithm = SunX509
> ssl.keystore.certificate.chain = null
> ssl.keystore.key = null
> ssl.keystore.location = null
> ssl.keystore.password = null
> ssl.keystore.type = JKS
> ssl.protocol = TLSv1.3
> ssl.provider = null
> ssl.secure.random.implementation = null
> ssl.trustmanager.algorithm = PKIX
> ssl.truststore.certificates = null
> ssl.truststore.location = null
> ssl.truststore.password = null
> ssl.truststore.type = JKS
> value.deserializer = class
> org.apache.kafka.common.serialization.StringDeserializer
>
> 2025-01-16 08:46:56,308 WARN
> (org.apache.kafka.clients.consumer.KafkaConsumer:688)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Support for using the
> empty group id by consumers is deprecated and will be removed in the next
> major release.
> 2025-01-16 08:46:56,330 INFO
> (org.apache.kafka.common.utils.AppInfoParser:119) [http-nio-8080-exec-3] -
> Kafka version: 3.6.1
> 2025-01-16 08:46:56,330 INFO
> (org.apache.kafka.common.utils.AppInfoParser:120) [http-nio-8080-exec-3] -
> Kafka commitId: 5e3c2b738d253ff5
> 2025-01-16 08:46:56,330 INFO
> (org.apache.kafka.common.utils.AppInfoParser:121) [http-nio-8080-exec-3] -
> Kafka startTimeMs: 1737046016330
> 2025-01-16 08:46:56,330 INFO
> (org.apache.kafka.clients.consumer.KafkaConsumer:1067)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Assigned to
> partition(s): ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
> 2025-01-16 08:46:56,330 INFO
> (org.apache.kafka.clients.consumer.KafkaConsumer:1067)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Assigned to
> partition(s): ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
> 2025-01-16 08:46:56,832 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:95) [http-nio-8080-exec-3] -
> EarliestOffset :0 LatestOffset :16136566 Time :502 ms
> 2025-01-16 08:46:56,832 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:97) [http-nio-8080-exec-3] -
> Latest & Earliest offset in Kafka for topic:
> ORG_CHANGE_EVENT_YS1_18_OCT_2024 & partition: 0 Latest is 16136566 Earliest
> is 0 and in user input is 16130077
> 2025-01-16 08:46:56,832 INFO
> (org.apache.kafka.clients.consumer.KafkaConsumer:1564)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Seeking to offset
> 16130077 for partition ORG_CHANGE_EVENT_YS1_18_OCT_2024-0
> 2025-01-16 08:46:57,157 INFO (org.apache.kafka.clients.Metadata:287)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Cluster ID:
> WyVVmQEwTbeRp73DYJgk7Q
> 2025-01-16 08:46:58,741 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:46:58,823 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:46:58,898 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:46:58,964 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :188
> 2025-01-16 08:47:00,243 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:00,316 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:00,383 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:00,455 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :235
> 2025-01-16 08:47:01,735 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:01,799 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:01,966 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:02,035 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :294
> 2025-01-16 08:47:02,912 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:02,981 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :500
> 2025-01-16 08:47:03,050 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :272
> 2025-01-16 08:47:05,087 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :0
> 2025-01-16 08:47:08,088 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :0
> 2025-01-16 08:47:11,088 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:114) [http-nio-8080-exec-3]
> - Poll Records Count :0
> 2025-01-16 08:47:11,088 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:157) [http-nio-8080-exec-3]
> - No messages were found in 3 successive fetches. Stopping the consume
> process here.
> 2025-01-16 08:47:11,088 INFO
> (com.test.cr.kafka.services.TestKafkaConsumer:179) [http-nio-8080-exec-3]
> - Summary ::: Total number of messages read from offset 16130077 to
> 16136566 are - 6489. Count of filtered results - 54
> 2025-01-16 08:47:11,088 INFO
> (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:1025)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Resetting generation
> and member id due to: consumer pro-actively leaving the group
> 2025-01-16 08:47:11,089 INFO
> (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:1072)
> [http-nio-8080-exec-3] - [Consumer
> clientId=crt_consumer_http-nio-8080-exec-3, groupId=] Request joining group
> due to: consumer pro-actively leaving the group
> 2025-01-16 08:47:11,138 INFO
> (org.apache.kafka.common.metrics.Metrics:693) [http-nio-8080-exec-3] -
> Metrics scheduler closed
> 2025-01-16 08:47:11,139 INFO
> (org.apache.kafka.common.metrics.Metrics:697) [http-nio-8080-exec-3] -
> Closing reporter org.apache.kafka.common.metrics.JmxReporter
> 2025-01-16 08:47:11,139 INFO
> (org.apache.kafka.common.metrics.Metrics:703) [http-nio-8080-exec-3] -
> Metrics reporters closed
> 2025-01-16 08:47:11,140 INFO
> (org.apache.kafka.common.utils.AppInfoParser:83) [http-nio-8080-exec-3] -
> App info kafka.consumer for crt_consumer_http-nio-8080-exec-3 unregistered
> 2025-01-16 08:47:11,159 INFO
> (com.cisco.cr.kafka.controller.EventsController:156)
> [http-nio-8080-exec-3] - Consume process end.
>
> New code:Here we are using KafkaConsumer API with SASL authentication
>
> public List<JSONObject> consume(long offset, String topicName,int
> partition,CEConsumeRequest inputReq) throws CustomException {
> List<JSONObject> msglist = new ArrayList<JSONObject>();
>
> int waitTime = Integer.valueOf(Config
> .getConsumerPropValue("pull.wait.time"));
> int limit =
> Integer.valueOf(Config.getConsumerPropValue("pull.size.limit"));
> int emptyLoopLimit = Integer.valueOf(Config
> .getConsumerPropValue("pull.empty.loop.limit"));
> int fetchSize = Integer.valueOf(Config
> .getConsumerPropValue("pull.each.fetch.size"));
>
> log.info(
> "Consume messages Topic: {} & partition: {}",
> topicName, partition);
>
> TopicPartition topicPartition = new TopicPartition(topicName, partition);
>
> long readOffset = offset;
> clientName = "crt_consumer_" + Thread.currentThread().getName();
> try (KafkaConsumer<String,String> consumer =
> KafkaConsumerFactory.createConsumer(clientName,fetchSize,topicPartition)){
> consumer.assign(Collections.singletonList(topicPartition));
>
> long OffsetStartTime = System.currentTimeMillis();
> long OffsetEndTime= System.currentTimeMillis();
> long kafkaEarliestOffset=0;
> long latestOffset=0;
> Admin adminClient = KafkaConsumerFactory.getAdminClient();
> try {
> kafkaEarliestOffset = adminClient.listOffsets(
> Map.of(topicPartition, OffsetSpec.earliest())
> ).all().get().get(topicPartition).offset();
> latestOffset = adminClient.listOffsets(
> Map.of(topicPartition, OffsetSpec.latest())
> ).all().get().get(topicPartition).offset();
> OffsetEndTime = System.currentTimeMillis();
> } catch (Exception e) {
> e.printStackTrace();
> }
> log.info("EarliestOffset :{} LatestOffset :{} Time :{}
> ms",kafkaEarliestOffset,latestOffset,(OffsetEndTime - OffsetStartTime));
>
> log.info(
> "Latest & Earliest offset in Kafka for topic: {} & partition: {}
> Latest is {} Earliest is {} and in user input is {}",
> topicName, partition, latestOffset, kafkaEarliestOffset,readOffset);
>
> if (readOffset == 0 && readOffset < kafkaEarliestOffset) {
> log.warn("Resetting the offset to earliest available offset in kafka.");
> readOffset = kafkaEarliestOffset;
> }
>
> boolean end = false;
> long startTime = System.currentTimeMillis();
> long endTime = System.currentTimeMillis();
> int emptyFetchCount = 0;
> consumer.seek(topicPartition, offset);
> do {
> JSONObject obj = null;
> ConsumerRecords < String, String > records =
> consumer.poll(Duration.ofMillis(2000));
> log.info("Poll Records Count :{}",records.count());
> for (ConsumerRecord< String, String > consumerRecord: records) {
> long currentOffset = consumerRecord.offset();
> if (currentOffset < readOffset) {
> log.warn("Found an old offset: {}, Expecting: {}",
> currentOffset, readOffset);
> continue;
> }
> String message = consumerRecord.value();
> log.debug(
> "client name : {} , Offset is : {} , Message is : {} ",
> clientName, readOffset,
> message);
> CONSUME_LOG.debug(topicName + "\t" + partition + "\t"
> + String.valueOf(currentOffset));
> obj = new JSONObject(message);
> if (!inputReq.getApplicationArea().isSubscription()
> || (obj.has("SUBSCRIBER_NAMES") && obj.getString(
> "SUBSCRIBER_NAMES").contains(
> inputReq.getSender().getReferenceID()))) {
> if (inputReq.getApplicationArea().getChangeEventsType() == null
> || (obj.has("NOTIFICATION_EVENT_NAME") && inputReq
> .getApplicationArea()
> .getChangeEventsType()
> .contains(
> obj.getString("NOTIFICATION_EVENT_NAME")))) {
> msglist.add(obj);
> }
> }
> readOffset = currentOffset + 1;
> }
> endTime = System.currentTimeMillis();
> 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;
> //consumer.commitSync();
> } else if (records.isEmpty()) {
> emptyFetchCount++;
> try {
> if(emptyFetchCount == emptyLoopLimit) {
> log.info("No messages were found in 3 successive fetches. Stopping the
> consume process here.");
> end = true;
> } else {
> Thread.sleep(1000);
> }
> } catch (InterruptedException ie) {
> CONSUME_LOG.warn(ie.getMessage(), ie);
> }
> } else {
> emptyFetchCount = 0;
> }
> } while (!end);
>
>
> The new kafka consumer API is very slow..i have tried lot of settings like
> increasing MAX_POLL_RECORDS_CONFIG,FETCH_MIN_BYTES_CONFIG etc..but no use.
>
> Is this time expected with the new KafkaConsumer API.How to improve my
> timings.
>
> please help on this.
>
> Regards,
> Giri
>