Kafka Connector

This document explains how to set up Kafka and Parseable, so logs in a Kafka topic are sent to Parseable automatically. We assume here that you'll setup the log agent to send logs to relevant Kafka topic.

Introduction

While Kafka and Parseable can be deployed in any environment, we're using Docker to document the deployment steps here. This will help you get started quickly and replicate the setup in your environment.

To set up Kafka, follow this guide.

Setup

Please ensure that Docker is installed on your machine.

Before starting the Parseable server, ensure that you have set the following environment variables-

• P_KAFKA_TOPIC - The topic that the Parseable server should subscribe to

• P_KAFKA_GROUP - The consumer group that the Parseable server is a part of

• P_KAFKA_HOST - The address of the Kafka broker

Following is a list of optional environment variables that Parseable accepts-

• P_KAFKA_CLIENT_ID - Unique ID for the Kafka client

• P_KAFKA_SECURITY_PROTOCOL - The security protocol to use

• P_KAFKA_PARTITIONS -

Then run the following commands to start the Parseable server.

Copy
docker run -p 8000:8000 \
  containers.parseable.com/parseable/parseable:edge \
  parseable local-store

Kafka Connect

Kafka Connect is a framework for connecting Kafka with external systems such as databases, key-value stores, search indexes, and file systems. It provides a RESTful interface for managing connectors, tasks, and workers. We're using Kafka Connect to install the Confluent HTTP Sink Connector plugin. This plugin will send logs from Kafka topic to Parseable.

Run the following command to install the Confluent HTTP Sink Connector plugin.

Copycurl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
    "name": "parseable-sink",
    "config": {
        "topics": "kafkademo",
        "tasks.max": "1",
        "connector.class": "io.confluent.connect.http.HttpSinkConnector",
        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
        "key.converter.schemas.enable": "false",
        "value.converter": "org.apache.kafka.connect.storage.StringConverter",
        "value.converter.schemas.enable": "false",
        "reporter.bootstrap.servers": "broker:29092",
        "reporter.result.topic.name": "success-responses",
        "reporter.result.topic.replication.factor": "1",
        "reporter.error.topic.name": "error-responses",
        "reporter.error.topic.replication.factor": "1",
        "confluent.topic.bootstrap.servers": "broker:29092",
        "confluent.topic.replication.factor": "1",
        "http.api.url": "http://parseable:8000/api/v1/ingest",
        "request.method": "POST",
        "headers":"X-P-Stream:kafkademo",
        "auth.type": "BASIC",
        "connection.user": "admin",
        "connection.password": "admin"
    }
}'

Sending data to Kafka topic

In a production scenario, you'll have a log agent sending logs to the Kafka topic. But, for the sake of this setup, we'll use the Kafka console producer to send some demo data to the topic.

Exec into the Kafka container and list the topics. You should see all the topics.

Copydocker exec -it broker /bin/bash
kafka-topics --list --bootstrap-server localhost:9092

Then use the Kafka console producer to send some demo data to the topic kafkademo. From within the Kafka container, run the following command.

Copykafka-console-producer --bootstrap-server localhost:9092 --topic kafkademo

You should see a prompt > to enter the data. Enter the following JSON objects one by one.

Copy{"reporterId": 8824, "reportId": 10000, "content": "Was argued independent 2002 film, The Slaughter Rule.", "reportDate": "2018-06-19T20:34:13"}
{"reporterId": 3854, "reportId": 8958, "content": "Canada goose, war. Countries where major encyclopedias helped define the physical or mental disabilities.", "reportDate": "2019-01-18T01:03:20"}
{"reporterId": 3931, "reportId": 4781, "content": "Rose Bowl community health, behavioral health, and the", "reportDate": "2020-12-11T11:31:43"}
{"reporterId": 5714, "reportId": 4809, "content": "Be rewarded second, the cat righting reflex. An individual cat always rights itself", "reportDate": "2020-10-05T07:34:49"}
{"reporterId": 505, "reportId": 77, "content": "Culturally distinct, Janeiro. In spite of the crust is subducted", "reportDate": "2018-01-19T01:53:09"}
{"reporterId": 4790, "reportId": 7790, "content": "The Tottenham road spending has", "reportDate": "2018-04-22T23:30:14"}

Now, to check if Parseable received the data, login to the Parseable UI on http://localhost:8000. Then select the kafkademo stream from the dropdown. You should see the data in the UI.