Skip to main content

Kafka

The Kafka module allows you to broadcast information on a Kafka bus. For example, you can use it to push alert or Situation data to a data lake via Kafka.

Load the module

The Kafka Moobot module is available to load into any standard Moobot. To load it, define a new global variable at the top of the Moobot Javascript file. For example:

var kafka = MooBot.loadModule('Kafka');

Method reference

The Kafka module uses the following methods.

connect

Establishes a connection to one or more Kafka brokers with defined connection properties.

Request arguments

The method takes the following arguments.

Name

Type

Required

Description

properties

Object

Yes

A JavaScript object containing connection properties. See Connection properties below.

Connection properties

The properties object can include the following keys:

Name

Type

Required

Description

servers

List

Yes

Specifies the Kafka broker or brokers to connect to.

connection_id

String

Yes

Name of the connection.

use_ssl

Boolean

Yes

Whether to connect using SSL.

ssl_truststore_location

String

No

Path to the SSL truststore file.

ssl_truststore_password

String

No

Password for the SSL truststore file.

ssl_truststore_encrypted_password

String

No

Encrypted password for the SSL truststore file.

ssl_key_password

String

No

Password for the SSL certificate.

ssl_key_encrypted_password

String

No

Encrypted password for the SSL certificate.

ssl_endpoint_identification_algorithm

String

No

Endpoint identification algorithm to validate the server hostname.

compression_codec

String

No

Compression codec to use. One of:

  • none

  • gzip

  • lz4

  • snappy

use_sasl

Boolean

No

Whether to connect using SASL.

sasl_mechanism

String

No

SASL mechanism to use. If you set a SASL mechanism you must set use_sasl to true. One of:

  • PLAIN

  • SCRAM-SHA-256

  • SCRAM-SHA-512

  • OAUTHBEARER

Choose from PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER.

security_protocol

String

No

SASL protocol to use. If you set a SASL protocol you must set use_sasl to true. One of:

  • SASL_SSL

  • SASL_PLAINTEXT

sasl_jaas_config

String

No

Credentials for JAAS authentication.

additional_properties

Various

No

Kafka consumer properties. Any properties you define here take priority over any configurations. See the Apache Kafka documentation for descriptions of these properties.

sasl_kerberos_service_name

String

No

Kerberos service name.

kerberos_conf_file_path

String

No

Path to the Kerberos configuration file.

kerberos_debug_log

Boolean

No

Whether to enable Kerberos authentication debug logs.

Response

The method returns the following parameter.

Type

Description

Object

A Java object containing connection details, depending on the requested connection properties. Returns null if no connection can be made.

Example

Example use of the connect method without SSL:

var conn = kafka.connect(
{
    servers: ["my.kafka.broker:9092", "my.other.kafka.broker:9092"],
    connection_id: "KafkaConnection",
    use_ssl: false
});

Example use of the connect method using SSL:

var conn = kafka.connect(
{
    servers: ["my.kafka.broker:9092", "my.other.kafka.broker:9092"],
    connection_id: "KafkaConnection",
    use_ssl: true,
    ssl_truststore_location: "/path/to/truststore",
    ssl_truststore_password: "test1234",
    ssl_keystore_location: "/path/to/keystore",
    ssl_keystore_password: "test1234",
    ssl_key_password: "test1234",
    ssl_endpoint_identification_algorithm: "",
    use_sasl: false
});

Example use of the connect method using SASL:

var conn = kafka.connect(
{
    servers: ["my.kafka.broker:9092", "my.other.kafka.broker:9092"],
    connection_id: "KafkaConnection",
    use_ssl: false,
    use_sasl: true,
    sasl_mechanism: "PLAIN",
    security_protocol: "SASL_PLAINTEXT",
    sasl_jaas_config: "org.apache.kafka.common.security.plain.PlainLoginModule required username=myuser password=mypassword;",
    additional_properties:{
       "sasl.login.refresh.window.factor": "0.8",
       "sasl.login.refresh.window.jitter": "0.05",
       "sasl.login.refresh.min.period.seconds": "60",
       "sasl.login.refresh.min.buffer.seconds": "300"
     },
    sasl_kerberos_service_name: "kafka",
    kerberos_conf_file_path : "/etc/krb5.conf",  
    kerberos_debug_log : true
});

send

Sends a message to the Kafka broker.

Request arguments

The method takes the following arguments.

Name

Type

Required

Description

topic

String

Yes

The Kafka topic to send the message on to.

key

String

No

The key to use for the message.

message

JSON

Yes

Message in one of the following formats:

  • Plain text

  • JSON Object payload

  • JSON Array payload

Response

The method returns the following parameter:

Type

Description

Boolean

Indicates whether the send operation was successful.

Example

Example use of the send method:

if (connection) 
{
    connection.send("myTopic", {test_field: "value"});
    connection.send("myTopic", "myKey", {test_field: "value"});
}

close

Closes the connection to the Kafka broker.

Request arguments

The method takes no arguments.

Response

None.

Examples

Example function using the Kafka module:

function sendToKafka(alert)
{
    var connection = kafka.connect({
        servers: ["my.kafka.broker:9092", "my.other.kafka.broker:9092"],
         connection_id: "KafkaConnection",
         use_ssl: false
    });

    if (connection){
        boolean success = connection.send("myTopic", {test_field: "value"});
         if (!success){
         logger.warning("Failed to send message to Kafka");
         }
    }
}
In this section: