consumers/bridge-consumer-group/instances/bridge-consumer/records Always returning HTTP 200 with [] array

[sreejesh-radhakrishnan-db]

I followed the steps from https://strimzi.io/docs/bridge/latest/#proc-bridge-subscribing-consumer-topics-bridge and when I run curl to consumers/bridge-consumer-group/instances/bridge-consumer/records I am getting empty array [].

Full command and log trace below:

       curl -X POST http://my-bridge-bridge-service.myapp-kafka:8080/consumers/bridge-consumer-group \
                                                    -H 'content-type: application/vnd.kafka.v2+json' \
                                                    -d '{
                                                      "name": "bridge-consumer",
                                                      "auto.offset.reset": "latest",
                                                      "format": "json",
                                                      "enable.auto.commit": true,
                                                      "fetch.min.bytes": 512,
                                                      "consumer.request.timeout.ms": 30000
                                                    }'

OUTPUT:

2023-02-01 09:50:58 INFO  createConsumer:85 - [1162488515] CREATE_CONSUMER Response:  statusCode = 200, message = OK
2023-02-01 09:50:58 INFO  HttpSinkBridgeEndpoint:143 - Created consumer bridge-consumer in group bridge-consumer-group
2023-02-01 09:50:58 INFO  AppInfoParser:121 - Kafka startTimeMs: 1675245058137
2023-02-01 09:50:58 INFO  AppInfoParser:120 - Kafka commitId: 839b886f9b732b15
2023-02-01 09:50:58 INFO  AppInfoParser:119 - Kafka version: 2.8.1

"	value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer"
"	ssl.truststore.type = PKCS12"
"	ssl.truststore.password = [hidden]"
"	ssl.truststore.location = /tmp/strimzi/bridge.truststore.p12"
"	ssl.truststore.certificates = null"
"	ssl.trustmanager.algorithm = PKIX"
"	ssl.secure.random.implementation = null"
"	ssl.provider = null"
"	ssl.protocol = TLSv1.3"
"	ssl.keystore.type = PKCS12"
"	ssl.keystore.password = [hidden]"
"	ssl.keystore.location = /tmp/strimzi/bridge.keystore.p12"
"	ssl.keystore.key = null"
"	ssl.keystore.certificate.chain = null"
"	ssl.keymanager.algorithm = SunX509"
"	ssl.key.password = null"
"	ssl.engine.factory.class = null"
"	ssl.endpoint.identification.algorithm = https"
"	ssl.enabled.protocols = [TLSv1.2, TLSv1.3]"
"	ssl.cipher.suites = null"
"	socket.connection.setup.timeout.ms = 10000"
"	socket.connection.setup.timeout.max.ms = 30000"
"	session.timeout.ms = 10000"
"	send.buffer.bytes = 131072"
"	security.providers = null"
"	security.protocol = SSL"
"	sasl.mechanism = GSSAPI"
"	sasl.login.refresh.window.jitter = 0.05"
"	sasl.login.refresh.window.factor = 0.8"
"	sasl.login.refresh.min.period.seconds = 60"
"	sasl.login.refresh.buffer.seconds = 300"
"	sasl.login.class = null"
"	sasl.login.callback.handler.class = null"
"	sasl.kerberos.ticket.renew.window.factor = 0.8"
"	sasl.kerberos.ticket.renew.jitter = 0.05"
"	sasl.kerberos.service.name = null"
"	sasl.kerberos.min.time.before.relogin = 60000"
"	sasl.kerberos.kinit.cmd = /usr/bin/kinit"
"	sasl.jaas.config = null"
"	sasl.client.callback.handler.class = null"
"	retry.backoff.ms = 100"
"	request.timeout.ms = 30000"
"	reconnect.backoff.ms = 50"
"	reconnect.backoff.max.ms = 1000"
"	receive.buffer.bytes = 65536"
"	partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]"
"	metrics.sample.window.ms = 30000"
"	metrics.recording.level = INFO"
"	metrics.num.samples = 2"
"	metric.reporters = []"
"	metadata.max.age.ms = 300000"
"	max.poll.records = 500"
"	max.poll.interval.ms = 300000"
"	max.partition.fetch.bytes = 1048576"
"	key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer"
"	isolation.level = read_uncommitted"
"	internal.throw.on.fetch.stable.offset.unsupported = false"
"	internal.leave.group.on.close = true"
"	interceptor.classes = []"
"	heartbeat.interval.ms = 3000"
"	group.instance.id = null"
"	group.id = bridge-consumer-group"
"	fetch.min.bytes = 512"
"	fetch.max.wait.ms = 500"
"	fetch.max.bytes = 52428800"
"	exclude.internal.topics = true"
"	enable.auto.commit = true"
"	default.api.timeout.ms = 60000"
"	connections.max.idle.ms = 540000"
"	client.rack = "
"	client.id = bridge-consumer"
"	client.dns.lookup = use_all_dns_ips"
"	check.crcs = true"
"	bootstrap.servers = [myapp-uat-kafka-bootstrap.myapp-kafka:9092]"
"	auto.offset.reset = latest"
"	auto.commit.interval.ms = 5000"
"	allow.auto.create.topics = true"
  ConsumerConfig:372 - ConsumerConfig values: 
  createConsumer:85 - [1162488515] CREATE_CONSUMER Request: from xxx.x.xxx.xxx:47292, method = POST, path = /consumers/bridge-consumer-group
 AppInfoParser:119 - Kafka version: 2.8.1
 AppInfoParser:120 - Kafka commitId: 839b886f9b732b15
 AppInfoParser:121 - Kafka startTimeMs: 1675245058137
 HttpSinkBridgeEndpoint:143 - Created consumer bridge-consumer in group bridge-consumer-group
 createConsumer:85 - [1162488515] CREATE_CONSUMER Response: statusCode = 200, message = OK
														{"instance_id":"bridge-consumer","base_uri":"http://my-bridge-bridge-service.myapp-kafka:8080/consumers/bridge-consumer-group/instances/bridge-consumer"}

COMMAND 2:

          curl -X POST http://my-bridge-bridge-service.myapp-kafka:8080/consumers/bridge-consumer-group/instances/bridge-consumer/subscription \
                                                      -H 'content-type: application/vnd.kafka.v2+json' \
                                                      -d '{
                                                        "topics": [
                                                            "le-event-log.proto"
                                                        ]
                                                    }'

OUTPUT AND LOG:

SUBSCRIBE Request: from xxx.x.xxx.xxx:51810, method = POST, path = /consumers/bridge-consumer-group/instances/bridge-consumer/subscription
Subscribe to topics [SinkTopicSubscription(topic=le-event-log.proto,partition=null,offset=null)]
 SUBSCRIBE Response: statusCode = 200, message = OK
 [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] Subscribed to topic(s): le-event-log.proto

COMMAND 3:

           curl   -X GET \
                   http://my-bridge-bridge-service.myapp-kafka:8080/consumers/bridge-consumer-group/instances/bridge-consumer/records \
                                                                     -H 'accept: application/vnd.kafka.json.v2+json'

 poll:85 - [504431900] POLL Request: from xxx.x.xxx.xxx:41396, method = GET, path = /consumers/bridge-consumer-group/instances/bridge-consumer/records
 poll:85 - [504431900] POLL Response: statusCode = 200, message = OK
 Metadata:279 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] Cluster ID: s4vro7U3RMSUKHQ5LdvcVw
 AbstractCoordinator:850 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] Discovered group coordinator myapp-uat-kafka-1.myapp-uat-kafka-brokers.myapp-kafka.svc:9092 (id: 2147483646 rack: null)
AbstractCoordinator:540 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] (Re-)joining group
poll:85 - [210260159] POLL Request: from xxx.x.xxx.xxx:54760, method = GET, path = /consumers/bridge-consumer-group/instances/bridge-consumer/records
 AbstractCoordinator:540 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] (Re-)joining group
 poll:85 - [210260159] POLL Response: statusCode = 200, message = OK
 AbstractCoordinator:611 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] JoinGroup failed: The coordinator is not aware of this member. Need to re-join the group. Sent generation was Generation{generationId=-1, memberId='bridge-consumer-c7bb1296-4044-4f9c-8eb8-5771b7740738', protocol='null'}
 AbstractCoordinator:470 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] Rebalance failed.
org.apache.kafka.common.errors.UnknownMemberIdException: The coordinator is not aware of this member.
INFO AbstractCoordinator:540 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] (Re-)joining group
INFO AbstractCoordinator:540 - [Consumer clientId=bridge-consumer, groupId=bridge-consumer-group] (Re-)joining group

and CURL output was :

% Total % Received % Xferd Average Speed Time Time Time Current
Dload Upload Total Spent Left Speed

0 0 0 0 0 0 0 0 --:--:-- --:--:-- --:--:-- 0
100 2 100 2 0 0 18 0 --:--:-- --:--:-- --:--:-- 18

[]

please let me know where I am going wrong?

[sreejesh-radhakrishnan-db]

KAFKA 3.1.0 and Operator 0.28

[ppatierno]

The HTTP bridge works as the Apache Kafka protocol.
A Kafka client subscribes to topic and then starts a polling loop. The subscribe operation doesn't do nothing more than adding topics to a local client list of subscriptions.
The initial poll requests (usually the first 2-3) are used to start the subscription process (join group and so on), so you could get empty records initially but then receiving messages.
Said that, I see you created the consumer with "auto.offset.reset": "latest" so it's not getting old messages but will get only new ones if you send them.