TimeoutException while sending data from Nifi to Kafka topic

[#1]

Hello All,

We are sending some data from Nifi to Kafka topic using PublishKafka processor. We are getting the following error:

PublishKafka Failed to send all messages for StandardFlowFileRecord to Kafka; routing to failure due to org.apache.kafka.common.errors.TimeoutException: Expiring 10 record(s) for test_topic:30011 ms has passed since batch creation plus linger time

We have tried increasing the linger time and request.timeout.ms but still facing the issue.

Any help would be appreciated.

[#2]

@R Patel

Could you please attach the broker server log covering the timestamp when you saw this error. Also what is the current value for linger time and request.timeout.ms?

Also please provide properties for PublishKafka processor. Are you seeing same issue if you publish to any other Kafka topic?

Thanks!

[#3]

Hi dbains,

Thanks for your answer.

As of now, the request.timeout.ms=30000 and linger.ms=1

PublishKafka processor properties are as following:

• Kafka Brokers: 10.2.2.10:6667
• Security Protocol: PLAINTEXT
• Topic Name: test_topic
• Delivery Guarantee: Best Effort
• Key Attribute Encoding: UTF-8 Encoded
• Max Request Size: 1 MB
• Acknowledgement Wait Time: 5 secs
• Max Metadata Wait Time: 5 secs

Yes, we see same issue for all the topics we tried.

Thanks.

[#4]

@dbains

Thanks for your answer.

As of now, the request.timeout.ms=30000 and linger.ms=1

PublishKafka processor properties are as following:

• Kafka Brokers: 10.2.2.10:6667
• Security Protocol: PLAINTEXT
• Topic Name: test_topic
• Delivery Guarantee: Best Effort
• Key Attribute Encoding: UTF-8 Encoded
• Max Request Size: 1 MB
• Acknowledgement Wait Time: 5 secs
• Max Metadata Wait Time: 5 secs

Yes, we see same issue for all the topics we tried.

Thanks.

[#5]

@R Patel

• Security Protocol: PLAINTEXT , looks like your Kafka cluster is not secure, could you please confirm?
• Could you please tell me your Kafka version as well.

Thanks!

[#6]

Hi @dbains

I am getting the same issue while pulling the data into PublishKafka procesor.

My conf for PublishKafka is :

• Kafka Brokers: ip:6667
• Security Protocol: PLAINTEXT
• Topic Name: test
• Delivery Guarantee: Best Effort
• Key Attribute Encoding: UTF-8 Encoded
• Max Request Size: 1 MB
• Acknowledgement Wait Time: 5 secs
• Max Metadata Wait Time: 5 secs

It throws the error Failed to update metadata 5000 ms.

Could you provide me any solution for this.

Thanks you.

[#7]

@Surendra Shringi

Hi,

Could you please try passing the FQDN of the kafka broker in the 'Kafka Brokers property of the processor?

Also may I know what version of Kafka are you trying to publish to?

Thanks!

[#8]

Hi @dbains, thanks for your quick response.

I am using PublishKafka_0_10_1.0.0 processor. and i have given FQDN in kafka property but still getting an error.

Do you have any demo template or any further help.

[#9]

Hi @Surendra Shringi,

May I know if you are connecting to a Kerberized Kafka or unsecure?

For template , kindly refer: https://community.hortonworks.com/articles/57262/integrating-apache-nifi-and-apache-kafka.html

Its an old article but it covers the basics in detail. Also, is Kafka in the same HDF cluster or are you using a separate Kafka cluster?

Thanks!

[#10]

We are using unsecure Kafka connection and not using MiNiFi, instead of that we are using MQTT.

this is our NiFi flow ..... MQTT --> PublishKafka --> ConsumeKafka --> PutCassandraQL.

What you suggest to use NiFi flow for below process ...

MQTT --> PublishKafka --> ConsumeKafka --> PutCassandraQL