Merge pull request #1321 from datanav/IS-18594

IS-18594: Improvements to kafka system, source and sink

Author: grove

hub/changelog.rst

@@ -1,6 +1,12 @@
 Changelog
 =========
 
+.. _changelog_2025-12-01:
+
+2025-12-01
+----------
+* Added schema registry and Avro support to the :ref:`Kafka system <kafka_system>`, :ref:`Kafka source <kafka_source>` and :ref:`Kafka sink <kafka_sink>`.
+
 .. _changelog_2025-10-10:
 
 2025-10-10

hub/documentation/service-configuration/pipes/configuration-sinks-kafka.rst

@@ -64,6 +64,12 @@ Properties
      - ``"json"``
      -
 
+   * - ``strategy``
+     - String
+     - Name of the strategy to use for the message. Allowed values are ``"key-value"`` and ``"value"``. The ``"key-value"`` will use the ``key`` property as the message key and ``value`` property as the message value. The ``"value"`` strategy will use the sink entity as the message value and the ``_id`` property as the message key.
+     - ``"key-value"``
+     -
+
    * - ``delivery_timeout_ms``
      - Integer
      - The time in milliseconds to await acknowledgement from the broker and the time allowed for retriable send failures.

hub/documentation/service-configuration/pipes/configuration-sources-kafka.rst

@@ -65,16 +65,22 @@ Properties
 
    * - ``key_deserializer``
      - String
-     - Name of the deserializer to use for the key. Allowed values are ``"bytes"``, ``"string"``, ``"json"``, ``"confluent_schema_json"``.
+     - Name of the deserializer to use for the key. Allowed values are ``"bytes"``, ``"string"``, ``"json"``, ``"schema_registry_avro"``.
      - ``"string"``
      -
 
    * - ``value_deserializer``
      - String
-     - Name of the deserializer to use for the value. Allowed values are ``"bytes"``, ``"string"``, ``"json"``, ``"confluent_schema_json"``.
+     - Name of the deserializer to use for the value. Allowed values are ``"bytes"``, ``"string"``, ``"json"``, ``"schema_registry_avro"``.
      - ``"json"``
      -
 
+   * - ``strategy``
+     - String
+     - Name of the strategy to use for the message. Allowed values are ``"key-value"`` and ``"value"``. The ``"key-value"`` will return the message key in the ``key`` property and the message value in the ``value`` property. The ``"value"`` strategy will return the message value as the source entity. The message ``_id`` will be generated from the message key if possible.
+     - ``"key-value"``
+     -
+
    * - ``strict``
      - Boolean
      - If the key or value cannot be deserialized then the pipe will fail if ``"strict"`` is ``true``. If ``false`` then pipe won't fail, but an invalid key will be stored in the ``"invalid_key"`` property and an invalid value will stored in the ``"invalid_value`` property. Note that if key serialization fails then the resulting entity won't have an ``"_id"`` property, so that must be dealt with in the pipe.

hub/documentation/service-configuration/systems/configuration-systems-kafka.rst

@@ -38,16 +38,33 @@ Properties
 
    * - ``sasl_username``
      - String
-     - Username to use when authentication against a SASL enabled Kafka cluster. If username is set, authentication will be performed.
+     - Username to use when authenticating against a SASL enabled Kafka cluster. If username is set, authentication will be performed.
        For Azure Event Hubs this property must be set to ``$ConnectionString`` and the connection string should be passed as the
        password.
      -
      - No
 
-
    * - ``sasl_password``
      - String
-     - Password to use when authentication against a SASL enabled Kafka cluster. For Azure Event Hubs this should be set to ``Endpoint=sb://[...]``.
+     - Password to use when authenticating against a SASL enabled Kafka cluster. For Azure Event Hubs this should be set to ``Endpoint=sb://[...]``.
+     -
+     - No
+
+   * - ``registry_url``
+     - String
+     - The URL of the schema registry endpoint.
      -
      - No
 
+   * - ``registry_username``
+     - String
+     - The username to use when authenticating against the schema registry.
+     -
+     - Required when ``registry_url`` is specified.
+
+   * - ``registry_password``
+     - String
+     - The password to use when authenticating against the schema registry.
+     -
+     - Required when ``registry_url`` is specified.
+