Skip to content
Integration pack for Apache Kafka message broker
Branch: master
Clone or download
Kami Merge pull request #9 from surfer190/master
Fixes no attribute '__dict__'
Latest commit 7e7ff93 May 14, 2019

README.md

Apache Kafka Integration Pack

This pack allows integration with Apache Kafka, high-throughput distributed messaging system. The KafkaGCPMessageSensor enables support for encoded GCP stackdriver formatted messages. Empowering Stackstorm with the ability to consume service provider level event information from Google Cloud. For more information on configuring stackdriver to PubSub to Kafka integration, please see https://github.com/GoogleCloudPlatform/pubsub/tree/master/kafka-connector

Actions

  • kafka.produce - Send message categorized by topic to Kafka hosts.

Sensors

KafkaMessageSensor

Connects to a Kafka broker, subscribing to various topics and dispatches triggers for each new message.

When receives new data, it emits:

  • trigger: kafka.new_message
  • payload:
    • topic - Category from which message was retrieved (string).
    • message - Message. JSON-serialized messages are converted to objects (object|string).
    • partition - Topic partition number message belongs to (integer).
    • offset - Consumer offset for current topic. Position of what has been consumed (integer).
    • key - Message's key, used only for keyed messages (string).

KafkaGCPMessageSensor

Connects to a Kafka broker, subscribing to various topics containing GCP stackdriver messages with base64 encoded payloads. The sensor decodes the payload and dispatches triggers for each new message.

When receives new data, it emits:

  • trigger: kafka.new_GCPmessage
  • payload:
    • topic - Category from which message was retrieved (string).
    • message - Message. JSON-serialized messages are converted to objects (object|string).
    • partition - Topic partition number message belongs to (integer).
    • offset - Consumer offset for current topic. Position of what has been consumed (integer).
    • key - Message's key, used only for keyed messages (string).

Configuration

  • hosts - Default hosts to send produce messages to in host:port format. Comma-separated for several hosts. (ex: localhost:9092)
  • client_id - Client ID to send with each message payload (default: st2-kafka-producer).
  • message_sensor:
    • hosts - Kafka hostname(s) to connect in host:port format. Comma-separated for several hosts. (ex: localhost:9092)
    • topics - Listen for new messages on these topics, (ex: ['test', 'meetings'])
    • group_id - Consumer group (default: st2-sensor-group)
    • client_id - Client ID to identify application making the request (default: st2-kafka-consumer).
  • gcp_message_sensor:
    • hosts - Kafka hostname(s) to connect in host:port format. Comma-separated for several hosts. (ex: localhost:9092)
    • topics - Listen for new messages with base64 encoded payloads on these topics, (ex: ['stackdriver', 'base64_messages'])
    • group_id - Consumer group (default: st2gcpgroup)
    • client_id - Client ID to identify application making the request (default: st2gcpclient).

Examples

Send message to Kafka queue:

# Publish message to `meetings` topic
st2 run kafka.produce topic=meetings message='StackStorm meets Apache Kafka'
# Send JSON-formatted message
st2 run kafka.produce topic=test message='{"menu": {"id": "file"}}'
You can’t perform that action at this time.