Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.


Panel

Table of Contents

Disclaimer

Your use of this download is governed by Stonebranch’s Terms of Use, which are available at  Stonebranch Integration Hub - Terms of Use.

Overview

Apache Kafka is an open-source distributed event streaming platform used by thousands of companies for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications. This Universal Extension is responsible for monitoring events (messages) from topics in Kafka, and consuming them based on filtering criteria via consumer group subscription.

...


Template NameExtension NameExtension Version
Apache Kafka: Event Monitorue-kafka-monitor1.1.0

Refer to Change Log for version history information.

Software Requirements

This integration requires a Universal Agent and a Python runtime to execute the Universal Task.

Software Requirements for Universal Template and Universal Task

Requires Python of version 3.7.  Tested with the Universal Agent bundled Python distribution (python version 3.7.6).

Software Requirements for Universal Agent

...

Supported Apache Kafka versions

 This Integration is tested on Kafka version 3.0.  Integration is expected to work with versions 2.0.0 onwards, however, this has not been tested.

Key Features

This Universal Extension provides the following main features:

...

Modifications of this integration, applied by users or customers, before or after import, might affect the supportability of this integration. For more information refer to Integration Modifications.

Configure Universal Task

...

Field

Input type

Default value

Type

Description

Action

Required

Monitor for events

Choice

The action performed upon the task execution.

Security Protocol

Required

PLAINTEXT

Choice

The Security protocol is used to communicate with Kafka brokers.

Valid values are:

  • PLAINTEXT

  • SASL_SSL

  • SSL

Bootstrap Servers

Required

-

Text

'host:port' string (or list of 'host:port' strings, separated by a comma) that the producer should contact to bootstrap initial cluster metadata.

This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request (more than one can be used, though, in case a server is down).

Example with two servers: 'host1:port1,host2:port2'.

SASL Mechanism

Optional

SCRAM–SHA–256

Choice

The Authentication mechanism when Security Protocol is configured for SASL_SSL.

Valid values are:

  • SCRAM–SHA–256: credentials are hashed with SHA–256 algorithm (requires a 32bit processor).

  • SCRAM–SHA–512: credentials are hashed with SHA–512 algorithm (requires a 64bit processor).

Required when Security Protocol is "SASL_SSL".

SASL User Credentials

Optional

-

Credentials

Credentials for SCRAM authentication.

They are comprised of:

  • Kafka host's username as "Runtime username".
  • Kafka host's password as "Runtime password".

Required when Security Protocol is "SASL_SSL".

SSL Hostname Check

Optional

true

Boolean

Flag to configure whether SSL handshake should verify that the certificate matches the broker's hostname.

Required when Security Protocol is "SASL_SSL" or "SSL".

CA Bundle Path

Optional

-

Text

Path and file name of the Certificate Authority (CA) file to use in certificate verification.

Used when it is required to locate the CA file if Security Protocol is configured for "SASL_SSL" or "SSL".

Client Certificate PathOptional-Text

Filepath of the Client's Certificate for Client authentication over SSL in PEM format.

Required when Security Protocol is "SSL".

Client Private Key PathOptional-Text

Filepath of the Client's private key for Client authentication over SSL in PEM format.

The private key can be either unencrypted or encrypted. In the case of an encrypted private key, the respective Client Private Key Password should be provided. 

Required when Security Protocol is "SSL".

Client Private Key PasswordOptional-Credential

In case the client's private key in Client Private Key Path is encrypted, the key required for decryption. The Credentials definition should be as follows.

  • Key Password as "Runtime Password".

Consumer Type

Required

Consumer Group

Choice

Type of Consumer to get messages from Kafka. Available options:

  • Consumer Group: a consumer is part of a consumer group.

Consumer Group

Required

-

Text

The unique name of the consumer group to join for dynamic partition assignment and to use for fetching and committing offsets.

Topic

Required

-

Dynamic Choice

Dynamic fetched list of topics to subscribe the consumer to.

The user can select the required topic from a drop-down list.

Client Id

Optional

-

Text

This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client.

The constructed client id seen on the server side is Client Id + "task instance id".

If Client Id is not populated, the default value used is "ue-kafka-monitor-#".

Example: "ue-kafka-monitor-#1635348654595881042JAYE7BKNPYVG3"

Start from

Required

Consumer Group Offset

Choice

Controls from which point the consumption will start.

Available option is:

  • Consumer Group Offset: start consuming from the committed offset of the group.

Key Deserializer

Required

String

Choice

Type of key deserialization.

Available options are:

  • Integer

  • String

Value Deserializer

Required

String

Choice

Type of value deserialization.

Available options are:

  • Integer

  • Float

  • String

  • JSON

Value Filter

Optional

None

Choice

Value operators to specify the criteria used to match records and stop consuming messages.

If Value Deserializer is configured for "Integer" or "Float", the available options are:

  • None (all messages are matched)

  • >= (greater than or equals to)

  • <= (less than or equals to)

  • = (equals to)

  • != (not equal to)

If Value Deserializer is configured for "String", the available options are:

  • None (all messages are matched)

  • Contains

  • Does Not Contain

  • Equals

  • Is Blank

  • Is Not Blank

If Value Deserializer is configured for "JSON", the available options are all that apply to the "Integer", "Float" and "String" Value Deserializer.

Value

Optional

-

Text

The Value on which the Value Filter applies to.

Value JSON Path

Optional

-

Text

The JSON path to locate the Value, in case Value Deserializer is configured to "JSON".

The JSON Path needs to resolve either to a number or to a string. If the JSON path results in a list of numbers or a list of strings, a Kafka message is matched if at least one element from the list matches the Value.

JSON Path syntax is based on jsonpath-ng python library. For examples, please refer to the official web site.

Show Advanced Settings

Required

False

Boolean

By checking this field, more fields are available for advanced configuration.

The advanced fields are: Partition Assignment Strategy, Session Timeout (ms), Auto Offset Reset, Request Timeout (ms), Heartbeat Interval (ms), Max Partition Fetch Bytes.

Partition Assignment Strategy

Optional

Range

Choice

Partition Assignment policies to distribute partition ownership amongst consumer instances when group management is used.

Available options are:

  • Range

  • Round Robin

Session Timeout (ms)

Optional

10000

Integer

Controls the time it takes to detect a consumer crash and stop sending heartbeats.

If more than Session Timeout milliseconds passes without the consumer sending a heartbeat to the group coordinator, it is considered dead and the group coordinator will trigger a rebalance of the consumer group to allocate partitions from the dead consumer to the other consumers in the group.

Auto Offset Reset

Optional

Latest

Choice

Controls the behavior of the consumer when it starts reading a partition for which it does not have a committed offset or the committed offset is invalid (usually because the consumer was down for so long that the record with that offset was already aged out of the broker).

Available options are:

  • Earliest which will move to the oldest available message.

  • Latest which will move to the most recent.

Request Timeout (ms)

Optional

305000

Integer

Controls the maximum amount of time the client will wait for the response of a request.

Heartbeat Interval (ms)

Optional

3000

Integer

The expected time in milliseconds between heartbeats to the consumer coordinator when using Kafka’s group management facilities.

The value must be set lower than Session Timeout (ms).

Max Partition Fetch Bytes

Optional

1048576

Integer

Controls the maximum number of bytes the server will return per partition.

This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition.

The default is 1MB.

...

Code Block
languagetext
linenumberstrue
{
   "exit_code": 0,
   "status_description": "SUCCESS: Successful Task execution",
   "changed": false,
   "invocation": {  ... }
   "result":{
      "client_id":"ue-kafka-monitor-#163578238979483604275PR4VPKK89D1",
      "event":{
         "key":1,
         "value":{
            "firstName":"John",
            "lastName":"doe",
            "age":26,
            "address":{
               "streetAddress":"naist street",
               "city":"Nara",
               "postalCode":"630-0192"
            },
            "phoneNumbers":[
               {
                  "type":"iPhone",
                  "number":"0123-4567-8888"
               },
               {
                  "type":"home",
                  "number":"0123-4567-8910"
               }
            ]
         },
         "headers":[],
         "partition":0,
         "offset":20,
         "timestamp":"2021-11-01T18:31:11.415000+00:00"
      }
   }
}

Matching a Kafka Event Kafka Event with Value Filter None

In this example, the task monitors for events and matches any Kafka event as the selected Value Filter is "None":

...

The Extension Output contains Attribute result. The result Attribute, as displayed below, is based on the Kafka response after the message is monitored and consumed successfully, based on Action "Monitor for events".

The Extension Output for this Universal Extension is in JSON format as described below.


Code Block
languagetext
linenumberstrue
{
    "exit_code": 0,
    "status_description": "SUCCESS: Successful Task execution",
    "changed": false,
    "invocation": {
        "extension": "ue-kafka-monitor",
        "version": "1.1.0",
        "fields": {
            "action": "Monitor for events",
            "auto_offset_reset": "Latest",
            "bootstrap_servers": [
                "kafka.plain.com:9092"
            ],
            "client_certificate_path": null,
            "client_id": "ue-kafka-monitor-#163578238979483604275PR4VPKK89D1",
            "client_private_key_path": null,
            "client_private_key_password": null,
            "consumer_type": "Consumer Group",
            "consumer_group": "float_group",
            "heartbeat_interval_ms": 3000,
            "key_deserializer": "String",
            "partition_assign_strategy": "RangePartitionAssignor",
            "request_timeout_ms": 305000,
            "start_from": "Consumer Group Offset",
            "security_protocol": "PLAINTEXT",
            "sasl_user_credentials_user": null,
            "sasl_user_credentials_password": null,
            "sasl_mechanism": "SCRAM-SHA-256",
            "ssl_check_hostname": true,
            "ssl_cafile": null,
            "session_timeout_ms": 10000,
            "topic": "test",
            "value": "3.14",
            "value_deserializer": "Float",
            "value_filter": "<=",
            "value_json_path": null
        }
    },
    "result": {
        "client_id": "ue-kafka-monitor-#163578238979483604275PR4VPKK89D1",
        "event": {
            "key": "float_group_key",
            "value": 3.14,
            "headers": [],
            "partition": 0,
            "offset": 46,
            "timestamp": "2023-05-08T11:38:38.026000+00:00"
        }
    }
}

...

STDOUT and STDERR provide additional information to User. The populated content can be changed in future versions of this extension without notice. Backward compatibility is not guaranteed.

Anchor
im
im

Integration Modifications

...