authorjapps / authorjapps/zerocode

Unable to consume the message from Kafka though message is present in the topic

Open
#425 6 comments 0 reactions 1 assignee Claimed by @nirmalchandra View on GitHub
kafka more-info-requested not-an-issue
Dominant language
Java
Stars
1k
Forks
453
Avg merge
7d 2h
Merged PRs (30d)
5

Description

Description
===
We have a service A which produces the message for Kafka and another service B which consumes and processes the messages. Using zerocode, We are able to load the messages in Kafka and other service is also able to process the same message. But when I try to consume those messages from Kafka using Zerocode, it is giving me size 0. Please find the logs below:
```bash
[0;39m2020-09-01 16:00:50,120 [main] INFO org.jsmart.zerocode.core.kafka.helper.KafkaConsumerHelper - 
---------------------------------------------------------
kafka.bootstrap.servers - .
---------------------------------------------------------
2020-09-01 16:00:50,121 [main] INFO org.jsmart.zerocode.core.kafka.client.BasicKafkaClient - brokers:, topicName:middleware_transaction_history_pg_data, operation:unload, requestJson:{"consumerLocalConfigs":{"showRecordsConsumed":true,"fileDumpTo":"target/temp/demo.txt","recordType":"JSON","maxNoOfRetryPollsOrTimeouts":5,"seek":"middleware_transaction_history_pg_data,0,0"}}
2020-09-01 16:00:50,140 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - 
### Kafka Consumer Effective configs:ConsumerLocalConfigs{recordType='JSON', fileDumpTo='target/temp/demo.txt', commitAsync=null, commitSync=null, showRecordsConsumed=true, maxNoOfRetryPollsOrTimeouts=5, pollingTime=null, seek=middleware_transaction_history_pg_data,0,0}

2020-09-01 16:00:50,182 [main] INFO org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [. there are 3 IPs]
check.crcs = true
client.dns.lookup = default
client.id =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = consumerGroup14
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
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]
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 = null
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.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.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

2020-09-01 16:00:50,407 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 2.1.0
2020-09-01 16:00:50,408 [main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : eec43959745f444f
2020-09-01 16:00:50,413 [main] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-1, groupId=consumerGroup14] Unsubscribed all topics or patterns and assigned partitions
2020-09-01 16:00:50,415 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - polling records - noOfTimeOuts reached : 0
2020-09-01 16:00:50,523 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - polling records - noOfTimeOuts reached : 1
2020-09-01 16:00:50,625 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - polling records - noOfTimeOuts reached : 2
2020-09-01 16:00:50,726 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - polling records - noOfTimeOuts reached : 3
2020-09-01 16:00:50,827 [main] INFO org.jsmart.zerocode.core.kafka.receive.KafkaReceiver - polling records - noOfTimeOuts reached : 4
2020-09-01 16:01:20,953 [main] INFO org.jsmart.zerocode.core.engine.validators.ZeroCodeValidatorImpl - Comparing results via LENIENT matchers
2020-09-01 16:01:20,962 [main] ERROR org.jsmart.zerocode.core.runner.StepNotificationHandler - Failed assertion during Scenario:CreateOrderThenPay, --> Step:Consume Kafka - bank, Details: Assertion jsonPath '$.size' with actual value '0' did not match the expected value '1'

2020-09-01 16:01:20,966 [main] ERROR org.jsmart.zerocode.core.runner.StepNotificationHandler - Assertion failed for :-

[CreateOrderThenPay]
|
|
+---Step --> [Consume Kafka - bank]

Failures:
---------
Assertion jsonPath '$.size' with actual value '0' did not match the expected value '1'
(See below 'Actual Vs Expected' to learn why this step failed)

2020-09-01 16:01:20,972 [main] WARN org.jsmart.zerocode.core.runner.ZeroCodeMultiStepsScenarioRunnerImpl - 
--------- TEST-STEP-CORRELATION-ID: 3ab7a9d1-f02f-4f93-b28f-bb296271efec ---------
*requestTimeStamp:2020-09-01T16:00:50.118
step:Consume Kafka
id:null
url:kafka-topic:middleware_transaction_history_pg_data12
method:UNLOAD
request:
{
"consumerLocalConfigs" : {
"showRecordsConsumed" : true,
"fileDumpTo" : "target/temp/demo.txt",
"recordType" : "JSON",
"maxNoOfRetryPollsOrTimeouts" : 5,
"seek" : "middleware_transaction_history_pg_data12,0,0"
}
}
--------- TEST-STEP-CORRELATION-ID: 3ab7a9d1-f02f-4f93-b28f-bb296271efec ---------
Response:
{
"records" : [ ],
"size" : 0
}
*responseTimeStamp:2020-09-01T16:01:20.952
*Response delay:30834.0 milli-secs
---------> Expected Response: <----------
Assumed Payload:
{
"size" : 1
}
Assertion Errors:
Assertion jsonPath '$.size' with actual value '0' did not match the expected value '1'


-done-
```

**Here are the Steps to unload Kafka**
```JSON
{
"name": "Consume Kafka - bank",
"url":"kafka-topic:middleware_transaction_history_pg_data12",
"operation": "unload",
"request": {
"consumerLocalConfigs": {
"showRecordsConsumed": true,
"fileDumpTo": "target/temp/demo.txt",
"recordType": "JSON",
"maxNoOfRetryPollsOrTimeouts": 5,
"seek": "middleware_transaction_history_pg_data12,0,0"
}
},
"assertions": {
"size": 1
}
}
```

**kafka_consumer.properties**
```properties
group.id=consumerGroup14
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
```

**Kafka_producer.properties**
```properties
client.id = test_producer_${LOCAL.DATE.TODAY:yyyy-MM-dd}
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.