Problem in Lambda for Java consumer listening to MSK topic

Question

I am observing a peculiar issue. I have built a Lambda function in Java 8; which polls a MSK topic using Kafka consumer api i.e. consumer.poll(5000) (I have tried with various timeout). There is a second Lambda function which is producer and sending messages to same topic. Both functions are attached to VPC in which MSK is in.
The producer working well. I can see the messages b running Kafka console consumer from an EC2. But the consumer Lambda does not work, it just gives timed out.

Only when I am running lambda producer, Kafka console consumer on the EC2 and Lambda consumer simultaneously, the consumer is getting some message !! To be precise, the producer sending 5 messages in a loop, the EC2 console consumer shows all 5, but the lambda consumer is showing 3rd or 4th message.

Why is this happening? What might be issue here, and How can I get the messages consistently in lambda consumer?

If anybody has a working code sample, I would be very grateful.

Thank you.

Further update: I have scheduled the consumer function, and then it gets all the events. I still have following questions - 1> Why is it not getting the messages when I manually trigger the function? 2> I tested a consumer function written in Python; that is not getting any messages too. Here is the python code:

def lambda_handler(event, context):
bootstrap_servers = ["<msk bootstrap>"]
topicName = '<mp-topic-name>'

consumer = KafkaConsumer (topicName, group_id = 'test',bootstrap_servers = bootstrap_servers,auto_offset_reset = 'earliest', consumer_timeout_ms=5000)

for message in consumer:
      consumer.commit()
      print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,message.offset, message.key,message.value))
    
KafkaConsumer.close()         
return ("Processed")

Code of the Java consumer; it is a plain Kafka client. From Lambda this is able to read messages when run in scheduled manner; but not when manually tested.

    try {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(topicName));

        context.getLogger().log("Subscribed to topic " + topicName);
        
        int i = 0;
        ConsumerRecords<String, String> records = consumer.poll(5000);          
        
        for (ConsumerRecord<String, String> record : records) {
            context.getLogger().log("Message:::  offset = "+record.offset()+", key = "+record.key()+", value = "+record.value()+"\n");
        }
        context.getLogger().log("After messages");
    } catch (Exception e) {
        e.printStackTrace();
        context.getLogger().log("Exception: "+e.getMessage());
    }

You post Python code in your question and tag it with java. What am I missing? — Abra
– Abra, Commented Jul 8, 2020 at 5:27
Hi @Abra, thanks for checking. As i updated, the java consumer is now working when i run it as scheduled Lambda. So my questions are -> 1) why the java consumer in Lambda not getting any messages when I run it manually? It is anyway reading all messages 2) python code posted is for yet another consumer, and this is unable to read the message from Kafka topic (whether run manually for scheduled). This defeats basic messaging. Why a python code will not be able to red from same topic? — sdas
– sdas, Commented Jul 9, 2020 at 3:11
Posted java code too above; it is just a plain Kafka consumer, nothing fancy. — sdas
– sdas, Commented Jul 9, 2020 at 3:14
Did you ever figure this out I've hit similar issue using c# stackoverflow.com/questions/63631372/… — sacha barber
– sacha barber, Commented Sep 1, 2020 at 8:18
@sachabarber ; no did not find any answer. temporarily switched to using SQS. — sdas
– sdas, Commented Sep 2, 2020 at 9:05

Misha Bruml · Accepted Answer · 2021-03-23 18:37:47Z

1

if only you'd have held on for another month or so! AWS now supports kafka event sources from both MSK and self-managed kafka clusters (even outside of AWS)

https://aws.amazon.com/blogs/compute/using-amazon-msk-as-an-event-source-for-aws-lambda/

https://aws.amazon.com/about-aws/whats-new/2020/12/aws-lambda-now-supports-self-managed-apache-kafka-as-an-event-source/

answered Mar 23, 2021 at 18:37

Misha Bruml

1396 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

sdas Over a year ago

Yes they did ; finally could not resist it any more :)

Collectives™ on Stack Overflow

Problem in Lambda for Java consumer listening to MSK topic

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related