feat:add kafka plugin

.github/workflows/check.yml

@@ -135,6 +135,7 @@ jobs:
               --ignore=tests/ft/app/test_ekuiper.py          \
               --ignore=tests/ft/app/test_mqtt.py             \
               --ignore=tests/ft/app/test_azure.py            \
+              --ignore=tests/ft/app/test_kafka.py            \
               --ignore=tests/ft/driver/test_modbus.py        \
               --ignore=tests/ft/metrics/test_metrics.py      \
               --ignore=tests/ft/login/test_launch.py         \
@@ -164,4 +165,73 @@ jobs:
           fail_ci_if_error: true
           token: ${{ secrets.CODECOV_TOKEN }}
           directory: ./cov_report/
-          files: ./cov_report/cov-${{ matrix.plugin }}.info
+          files: ./cov_report/cov-${{ matrix.plugin }}.info
+
+  pft_kafka:
+    runs-on: ubuntu-24.04
+    container: ghcr.io/neugates/build:x86_64-main
+    needs: [build]
+    services:
+      kafka:
+        image: apache/kafka:3.7.0
+        env:
+          KAFKA_NODE_ID: "1"
+          KAFKA_PROCESS_ROLES: broker,controller
+          KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
+          KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+          KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "1"
+          KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
+          CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
+        ports:
+          - 9092:9092
+        options: >-
+          --health-cmd "bash -c 'echo > /dev/tcp/localhost/9092'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+          --health-start-period 30s
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: recursive
+
+      - name: setup dependencies
+        run: |
+          python3 -m pip install --upgrade pip
+          pip install -U pytest requests confluent-kafka==2.3.0
+
+      - uses: actions/download-artifact@v4
+        with:
+          name: neuron-build
+          path: ./build
+
+      - name: function test
+        env:
+          KAFKA_BROKER: kafka:9092
+        run: |
+          chmod -R +x ./build/
+          cd build
+          cmake -DUSE_GCOV=1 .. -DCMAKE_TOOLCHAIN_FILE=../cmake/x86_64-linux-gnu.cmake
+          cd ..
+          ldconfig
+          pytest -s -v tests/ft/app/test_kafka.py
+
+      - name: create cov report
+        run: |
+            sudo apt-get update
+            sudo apt-get -y install lcov
+            mkdir -p cov_report
+            ./create_cov_report.sh kafka
+
+      - name: Upload coverage to Codecov
+        uses: codecov/codecov-action@v4
+        with:
+          verbose: true
+          fail_ci_if_error: true
+          token: ${{ secrets.CODECOV_TOKEN }}
+          directory: ./cov_report/
+          files: ./cov_report/cov-kafka.info

CMakeLists.txt

@@ -196,6 +196,7 @@ add_subdirectory(plugins/ekuiper)
 add_subdirectory(plugins/file)
 add_subdirectory(plugins/monitor)
 add_subdirectory(plugins/datalayers)
+add_subdirectory(plugins/kafka)
 
 add_subdirectory(simulator)
 

Install-dependencies.md

@@ -74,6 +74,13 @@ $ ./configure  --disable-protoc --enable-shared=no CFLAGS=-fPIC CXXFLAGS=-fPIC
 $ make && sudo make install
 ```
 
+[librdkafka](https://github.com/confluentinc/librdkafka.git)
+```shell
+$ git clone -b v2.6.1 https://github.com/confluentinc/librdkafka.git
+$ cd librdkafka && mkdir build && cd build
+$ cmake -DRDKAFKA_BUILD_STATIC=OFF -DRDKAFKA_BUILD_EXAMPLES=OFF -DRDKAFKA_BUILD_TESTS=OFF -DWITH_SASL=OFF -DWITH_ZSTD=OFF -DWITH_CURL=OFF -DCMAKE_BUILD_TYPE=Release .. && make && sudo make install
+```
+
 [libxml2](https://github.com/GNOME/libxml2.git)
 ```shell
 $ git clone -b v2.9.14 https://github.com/GNOME/libxml2.git

cmake/aarch64-linux-gnu.cmake

@@ -21,4 +21,5 @@ set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
 
-file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/librdkafka.so.1 DESTINATION /usr/local/lib)

cmake/arm-linux-gnueabi.cmake

@@ -22,3 +22,4 @@ set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
 
 file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/librdkafka.so.1 DESTINATION /usr/local/lib)

cmake/arm-linux-gnueabihf.cmake

@@ -21,4 +21,5 @@ set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
 
-file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/librdkafka.so.1 DESTINATION /usr/local/lib)

cmake/x86_64-linux-gnu.cmake

@@ -27,3 +27,4 @@ set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
 
 file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/librdkafka.so.1 DESTINATION /usr/local/lib)

default_plugins.json

@@ -6,6 +6,7 @@
 		"libplugin-ekuiper.so",
 		"libplugin-modbus-tcp.so",
 		"libplugin-modbus-rtu.so",
-		"libplugin-file.so"
+		"libplugin-file.so",
+		"libplugin-kafka.so"
 	]
 }

package-sdk.sh

@@ -59,15 +59,19 @@ cp build/plugins/schema/ekuiper.json \
     build/plugins/schema/mqtt.json \
     build/plugins/schema/modbus-tcp.json \
     build/plugins/schema/file.json \
+    build/plugins/schema/kafka.json \
     ${package_name}/plugins/schema/
 
 cp build/plugins/libplugin-ekuiper.so \
     build/plugins/libplugin-monitor.so \
     build/plugins/libplugin-mqtt.so \
     build/plugins/libplugin-modbus-tcp.so \
     build/plugins/libplugin-file.so \
+    build/plugins/libplugin-kafka.so \
     ${package_name}/plugins/
 
+cp /usr/local/lib/librdkafka.so.1 ${package_name}/lib
+
 tar czf ${package_name}-${arch}.tar.gz ${package_name}/
 ls ${package_name}
 rm -rf ${package_name}

plugins/kafka/CMakeLists.txt

@@ -0,0 +1,62 @@
+cmake_minimum_required(VERSION 3.12)
+
+project(plugin-kafka)
+
+set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
+
+file(COPY ${CMAKE_SOURCE_DIR}/plugins/kafka/kafka.json
+     DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
+
+# --- locate librdkafka -------------------------------------------------------
+# Prefer pkg-config (works for system install and staging sysroot).
+# Falls back to find_library for minimal environments.
+find_package(PkgConfig QUIET)
+if(PKG_CONFIG_FOUND)
+    pkg_check_modules(RDKAFKA IMPORTED_TARGET rdkafka)
+endif()
+
+if(NOT RDKAFKA_FOUND)
+    find_path(RDKAFKA_INCLUDE_DIR librdkafka/rdkafka.h
+              HINTS ${CMAKE_STAGING_PREFIX}/include
+                    /usr/local/include
+                    /usr/include)
+    find_library(RDKAFKA_LIBRARY rdkafka
+                 HINTS ${CMAKE_STAGING_PREFIX}/lib
+                       /usr/local/lib
+                       /usr/lib)
+    if(RDKAFKA_INCLUDE_DIR AND RDKAFKA_LIBRARY)
+        set(RDKAFKA_FOUND TRUE)
+        set(RDKAFKA_INCLUDE_DIRS ${RDKAFKA_INCLUDE_DIR})
+        set(RDKAFKA_LIBRARIES    ${RDKAFKA_LIBRARY})
+    else()
+        message(FATAL_ERROR
+            "librdkafka not found. Install librdkafka-dev or set "
+            "CMAKE_STAGING_PREFIX / PKG_CONFIG_PATH so that rdkafka.h "
+            "and librdkafka.so are discoverable.")
+    endif()
+endif()
+
+# --- plugin target ------------------------------------------------------------
+add_library(${PROJECT_NAME} SHARED
+    kafka_config.c
+    kafka_handle.c
+    kafka_plugin.c
+    kafka_plugin_intf.c
+)
+
+target_include_directories(${PROJECT_NAME} PRIVATE
+    ${CMAKE_SOURCE_DIR}/include/neuron
+    ${CMAKE_SOURCE_DIR}/plugins/kafka
+)
+
+if(TARGET PkgConfig::RDKAFKA)
+    target_link_libraries(${PROJECT_NAME} neuron-base PkgConfig::RDKAFKA)
+else()
+    target_include_directories(${PROJECT_NAME} PRIVATE ${RDKAFKA_INCLUDE_DIRS})
+    target_link_libraries(${PROJECT_NAME} neuron-base ${RDKAFKA_LIBRARIES})
+endif()
+
+target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT})
+
+# librdkafka header triggers -Wignored-qualifiers on some versions
+target_compile_options(${PROJECT_NAME} PRIVATE -Wno-ignored-qualifiers)