Kafka

The kafka node-type uses librdkafka to produce and consume event stream messages to/from kafka message brokers cluster.

Prerequisites

This node-type requires rdkafka (>= 1.5.0).

Implementation

The source code of the node-type is available here: https://github.com/VILLASframework/node/blob/master/lib/nodes/kafka.cpp

Configuration

	Format Object (object) or Format Name (string) (format_spec)
server required	string The bootstrap server {ip}:{port} of the Kafka message brokers cluster.
protocol	string Enum: "PLAINTEXT" "SASL_PLAINTEXT" "SASL_SSL" "SSL" The security protocol which is used for authentication with the Kafka cluster.
client_id required	string The Kafka client identifier.
	object
	object An object for configuring the SASL authentication against the broker. This setting is used if the protocol setting is on of SASL_PLAINTEXT or SASL_SSL.
	object (Input configuration (received by VILLASnode))
	object (Output configuration (sent out by VILLASnode))
timeout	number Default: 1 A timeout in seconds for the broker connection.
vectorize	integer Default: 1 This setting allows to send multiple samples in a single message to the destination nodes. The value of this setting determines how many samples will be combined into one packet.
	Array of Hook Object (object) or Hook Name (string) (hook_list)
builtin	boolean (Builtin hook functions) Default: true By default, each node and paths has a couple of default hooks attached to them. With this setting the attachment of built-in hooks can be disabled.

{

• "format": "villas.human",
• "server": "string",
• "protocol": "PLAINTEXT",
• "client_id": "string",
• "ssl": {

  • "ca": "string"

  },
• "sasl": {

  • "mechanisms": "string",
  • "username": "string",
  • "password": "string"

  },
• "in": {

  • "consume": "string",
  • "group_id": "string",
  • "signals": [

    • {

      • "name": "tap_position",
      • "type": "integer",
      • "init": 0

      },
    • {

      • "name": "voltage",
      • "type": "float",
      • "unit": "V",
      • "init": 230

      }

    ],
  • "vectorize": 1,
  • "hooks": [

    • "print",
    • {

      • "type": "limit_rate",
      • "rate": 1000

      }

    ]

  },
• "out": {

  • "produce": "string",
  • "vectorize": 0,
  • "hooks": [

    • "print",
    • {

      • "type": "limit_rate",
      • "rate": 1000

      }

    ]

  },
• "timeout": 1,
• "vectorize": 1,
• "hooks": [

  • "print",
  • {

    • "type": "limit_rate",
    • "rate": 1000

    }

  ],
• "builtin": true

}

Example

node/etc/examples/nodes/kafka.conf

nodes = {
    kafka_node = {
        type = "kafka"

        format = "json.kafka"

        server = "localhost:9094"
        protocol = "SASL_SSL"
        client_id = "villas-node"

        in = {
            consume = "test-topic"
            group_id = "villas-node"
        }

        out = {
            produce = "test-topic"
        }

        ssl = {
            ca = "/etc/ssl/certs/ca.pem"
        }

        sasl = {
            mechanisms = "SCRAM-SHA-512"
            username = "scram-sha-512-usr"
            password = "scram-sha-512-pwd"
        }
    }
}