Kafka sink with EdgeX foundry ekuiper

[mohammedimrankasab]

Hi

we are trying to add kafka as a sink with EdgeX foundry ekuiper engine using docker compose. As per the documentation https://ekuiper.org/docs/en/latest/guide/sinks/plugin/kafka.html we have configured the plugin with the exact same way mentioned in the docmentation.

we have used the pre-compiled kafka-plugin which is available publicly. ekuiper and kafka plugin version is 1.11.4

Once the setup is up and running we are encountering the issue on kafka side once the events are getting pushed from the EdgeX foundry. Below are the error logs. We have tried updating the default size for the message and socket to a big number like 400 MB but after that change a new error of OOMKilled is popping up.

Need expertise advice here if something is missing on the configuration or suggestion on how to proceed on this

[2024-02-07 07:22:05,839] WARN [SocketServer listenerType=ZK_BROKER, nodeId=1001] Unexpected error from /192.168.80.6 (channelId=192.168.80.9:9092-192.168.80.6:56246-0); closing connection (org.apache.kafka.common.network.Selector)
org.apache.kafka.common.network.InvalidReceiveException: Invalid receive (size = 369295616 larger than 104857600)
        at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:94)
        at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452)
        at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402)
        at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674)
        at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576)
        at org.apache.kafka.common.network.Selector.poll(Selector.java:481)
        at kafka.network.Processor.poll(SocketServer.scala:1056)
        at kafka.network.Processor.run(SocketServer.scala:960)
        at java.base/java.lang.Thread.run(Thread.java:840)

create a stream
{"sql":"create stream EdgeXStream () WITH ( FORMAT = "json", TYPE="edgex")"}

create a rule
{
"id": "kafka",
"sql": "SELECT * from EdgeXStream where value >0.7",
"actions": [
{
"kafka": {
"brokers": "kafka:9092",
"topic": "test_topic",
"saslAuthType": "none"
}
}
]
}

[ngjaying]

@Yisaer Could you please take a look?

[Yisaer]

Hi @mohammedimrankasab

please upgrade the eKuiper and kafka sink plugin to the latest version and try if the error still exsits

[mohammedimrankasab]

Hi @Yisaer

Yes, the latest version of the 1.12.4-alpine image works like a charm. Can you let us know from which version the fix for Edgex Foundry and kafka plugin is introduced?

Thanks