1 of 10

Kafka

This page describes how to connect the Lenses Agent to your Kafka brokers.

The Lenses Agent can connect to any Kafka cluster or service exposing the Apache Kafka APIs and supporting the authentication methods offered by Apache Kafka.

Apache Kafka

This page describes connecting the Lenses Agent to Apache Kafka.

A Kafka connection is required for the agent to start. You can connect to Kafka via:

Plaintext (no credentials an unencrypted)
SSL (no credentials an encrypted)
SASL Plaintext and SASL SSL

Plaintext

With PLAINTEXT, there's no encryption and no authentication when connecting to Kafka.

The only required fields are:

kafkaBootstrapServers - a list of bootstrap servers (brokers). It is recommended to add as many brokers (if available) as convenient to this list for fault tolerance.
protocol - depending on the protocol, other fields might be necessary (see examples for other protocols)

In following example JMX metrics for Kafka Brokers are configured too, assuming that all brokers expose their JMX metrics using the same port (9581), without SSL and authentication.

SSL

With SSL the connection to Kafka is encrypted. You can also uses SSL and certificates to authenticate users against Kafka.

A truststore (with password) might need to be set explicitly if the global truststore of the Agent does not include the Certificate Authority (CA) of the brokers.

If TLS is used for authentication to the brokers in addition to encryption-in-transit, a key store (with passwords) is required.

SASL Plaintext vs SASL SSL

There are 2 SASL-based protocols to access Kafka Brokers: SASL_SSL and SASL_PLAINTEXT. They both require SASL mechanism and JAAS Configuration values. What is different is:

The transport layer is encyrpted (SSL)
The SASL mechanism for authentication (PLAIN, AWS_MSK_IAM, GSSAPI).

In addition to this, there might be a keytab file required, depending on the SASL mechanism (for example when using GSSAPI mechanism, most often used for Kerberos).

To use Kerberos authentication, a Kerberos _Connection_ should be created beforehand.

When encryption-in-transit is used (with SASL_SSL), a trust store might need to be set explicitly if the global trust store of Lenses does not include the CA of the brokers.

SASL SSL

Mechanism PLAIN

Encrypted communication and basic username and password for authentication.

Mechanism GSSAPI

In order to use Kerberos authentication, a Kerberos Connection should be created beforehand.

SASL Plaintext

No SSL encrypted of communication, credentials communicated to Kafka in clear text.

Mechanism SCRAM-SHA-256

Mechanism SCRAM-SHA-512

Aiven

This page describes configuring Lenses to connect to Aiven.

Find your Service URI

From the Aiven, locate your Service URI and set it as the bootstrap servers.

Configure Provisioning

Set the following in the provisioning.yaml, replacing Service URI, username and password from your Aiven account.

provisioning.yaml

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_SSL://[Service URI]
    protocol: 
      value: SASL_SSL
    saslMechanism: 
      value: SCRAM-SHA-256
    saslJaasConfig: 
      value: |
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="[your-username]"
        password="[your-password]";

AWS MSK

This page describes connection the Lenses Agent to a AWS MSK cluster.

It is recommended to install the Agent on an EC2 instance or with EKS in the same VPC as your MSK cluster. The Agent can be installed and preconfigured via the AWS Marketplace.

Open network connectivity

Edit the AWS MSK security group in the AWS Console and add the IP address of your Agent installation.

Enable Open Monitoring

If you want to have the Agent collect JMX metrics you have to enable Open Monitoring on your MSK cluster. Follow the AWS guide here.

Select your MSK endpoint

Depending on your MSK cluster, select the endpoint and protocol you want to connect with.

It is not recommended to use Plaintext for secure environments. For these environments use TLS or IAM.

When the Agent is running inside AWS and is connecting to an Amazon’s Managed Kafka (MSK) instance, IAM can be used for authentication.

Configure Provisioning

provisioning.yaml

kafka:
- name: Kafka
  version: 1
  tags: ["optional-tag"]
  configuration:
    kafkaBootstrapServers:
      value:
       - SASL_SSL://your.kafka.broker.0:9098
       - SASL_SSL://your.kafka.broker.1:9098
    protocol: SASL_SSL
    saslMechanism: 
      value: AWS_MSK_IAM
    saslJaasConfig:
      value: software.amazon.msk.auth.iam.IAMLoginModule required;
    additionalProperties:
      value:
        sasl.client.callback.handler.class: "software.amazon.msk.auth.iam.IAMClientCallbackHandler"

AWS MSK Serverless

This page describes how to connect Lenses to an Amazon MSK Serverless cluster.

It is recommended to install the Agent on an EC2 instance or with EKS in the same VPC as your MSK Serverless cluster.

Security Groups

Enable communications between the Agent & the Amazon MSK Serverless cluster by opening the Amazon MSK Serverless cluster's security group in the AWS Console and add the IP address of your Agent installation.

IAM Policy

To authenticate the Agent & access resources within our MSK Serverless cluster, we'll need to create an IAM policy and apply that to the resource (EC2, EKS cluster, etc) running the Agent service. here is an example IAM policy with sufficient permissions which you can associate with the relevant IAM role:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:AlterCluster",
                "kafka-cluster:DescribeCluster"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:CreateTopic",
                "kafka-cluster:WriteData",
                "kafka-cluster:ReadData"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:topic/[cluster_name]/[cluster_uuid]/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:group/[cluster_name]/[cluster_uuid]/*"
        }
    ]
}

MSK Serverless IAM to be used after cluster creation. Update this IAM policy with the relevant ARN.

Select your MSK endpoint

Click your MSK Serverless Cluster in the MSK console and select View Client Information page to check the bootstrap server endpoint.

Configure Provisioning

provisioning.yaml

kafka:
  tags: ["optional-tag"]
  name: Kafka
  configuration:
    kafkaBootstrapServers:
      value:
       - SASL_SSL://your.kafka.broker.0:9098
       - SASL_SSL://your.kafka.broker.1:9098
    protocol: SASL_SSL
    saslMechanism: 
      value: AWS_MSK_IAM
    saslJaasConfig:
      value: software.amazon.msk.auth.iam.IAMLoginModule required;
    additionalProperties:
      value:
        sasl.client.callback.handler.class: "software.amazon.msk.auth.iam.IAMClientCallbackHandle

SQL Processor Configurations

To enable the creation of SQL Processors that create consumer groups, you need to add the following statement in your IAM policy:

{
  "Action": [
    "kafka-cluster:*Topic*",
    "kafka-cluster:WriteData",
    "kafka-cluster:ReadData"
  ],
  "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
}

Update the placeholders in the IAM policy based on the relevant MSK Serverless cluster ARN.

To integrate with the AWS Glue Schema Registry, you also need to add the following statement for the registries and schemas in your IAM policy:

{
  "Action": [
    "kafka-cluster:*Group*"
  ],
  "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
}

Update the placeholders in the IAM policy based on the relevant MSK Serverless cluster ARN.

To integrate with the AWS Glue Schema Registry, you also need to modify the security policy for the registry and schemas, which results in additional functions within it:

{
  "Action": [
    "glue:DeregisterDataPreview",
    "glue:ListRegistries",
    "glue:CreateRegistry",
    "glue:RegisterSchemaVersion",
    "glue:GetRegistry",
    "glue:UpdateRegistry",
    "glue:ListSchemas",
    "glue:DeleteRegistry",
    "glue:GetSchema",
    "glue:CreateSchema",
    "glue:ListSchemaVersions",
    "glue:GetSchemaVersion",
    "glue:UpdateSchema",
    "glue:DeleteSchemaVersions"
  ],
  "Resource": [
    "arn:aws:glue:[region]:[aws_account_id]:registry/*",
    "arn:aws:glue:[region]:[aws_account_id]:schema/*"
  ]
}

More details about how IAM works with MSK Serverless can be found in the documentation: MSK Serverless

Limitations

When using the Agent with MSK Serverless:

The agent does not receive Prometheus-compatible metrics from the brokers because they are not exported outside of CloudWatch.
The agent does not configure quotas and ACLs because MSK Serveless does not allow this.

Azure EventHubs

This page describes connection Lenses to Azure EventHubs.

Create a Data Integration API key

Add a shared access policy
- Navigate to your Event Hub resource and select Shared access policies in the Settings section.
- Select + Add shared access policy, give a name, and check all boxes for the permissions (Manage, Send, Listen)
Once the policy is created, obtain the Primary Connection String, by clicking the policy and copying the connection string. The connection string will be used as a JAAS password to connect to Kafka.
The bootstrap broker [YOUR_EVENT_HUBS_NAMESPACE].servicebus.windows.net:9093

Configure Provisioning

Set the following in the provisioning.yaml

First set environment variable

Note that "\" at "$ConnectionString" is set additionally to escape the $ sign.

terminal

export SASL_JAAS_CONFIG=org.apache.kafka.common.security.plain.PlainLoginModule required username="\$ConnectionString" password="Endpoint=sb://[SB_URL]/;SharedAccessKeyName=[KEY_NAME];SharedAccessKey=[ACCESS_KEY]";

provision.yaml

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
        - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
    saslJaasConfig:
      value: '${SASL_JAAS_CONFIG}'
    saslMechanism:
      value: PLAIN
    protocol:
      value: SASL_SSL

provision.yaml

connections:
  kafka:
  - name: Kafka
    version: 1
    tags: [my-tag]
    configuration:
      kafkaBootstrapServers:
        value:
          - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
          - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
      saslJaasConfig:
        value: org.apache.kafka.common.security.plain.PlainLoginModule required username="\$ConnectionString" password="Endpoint=sb://[SB_URL]/;SharedAccessKeyName=[KEY_NAME];SharedAccessKey=[ACCESS_KEY]";
      saslMechanism:
        value: PLAIN
      protocol:
        value: SASL_SSL

Azure HDInsight

This page describes connection Lenses to a Azure HDInsight cluster.

Find your Kafka endpoints

In our Azure Portal, go to Dashboards->Ambari home. Go to Kafka->Configs->Kafka Broker->Kafka Broker hosts

Optionally find your Zookeeper endpoints

Optionally get the Zookeeper endpoints: Go to Zookeeper->Configs->Zookeeper Server->Zookeeper Server hosts.

Configure Provisioning

See Apache Kafka for different configurations for your security protocols.

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - PLAINTEXT://your.kafka.broker.0:9092
        - PLAINTEXT://your.kafka.broker.1:9092
    protocol: 
      value: PLAINTEXT
    # all metrics properties are optional
    metricsPort: 
      value: 9581
    metricsType: 
      value: JMX
    metricsSsl: 
      value: falseSSL

Confluent Cloud

This page describes configuring Lenses to connect to Confluent Cloud.

For Confluent Platform see Apache Kafka.

Create a Data Integration API key

From Data integration API keys, select Create Key.
For this guide select Global access to get your API Key and API Secret Key.
Go to Cluster Settings to get your Bootstrap Server.

Configure Provisioning

Set the following in the provisioning.yaml

provisioning.yaml

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
        - SASL_SSL://[YOUR_BOOTSTRAP_SERVER]
    protocol: 
      value: SASL_SSL
    saslMechanism: 
      value: PLAIN
    saslJaasConfig:
      value: |
        org.apache.kafka.common.security.plain.PlainLoginModule required 
        username="[YOUR_API_KEY]" 
        password="[YOUR_API_KEY_SECRET]"

Confluent Platform

This page describes configuring Lenses to connect to Confluent Platform.

For Confluent Platform see Apache Kafka.

IBM Event Streams

This page describes how to connect Lenses to IBM Event Streams.

IBM Event Streams requires a replication factor of 3. Ensure you set the replication factor accordingly for Lenses internal topics.

See .

Find your bootstrap endpoints

From the IBM Cloud console, locate your bootstrap_endpoints, for the service credentials you want to connect with.

Configure Provisioning

Set the following in the provisioning.yaml:

Use "token" as the username in the Jaas Config. Set the password as your API KEY from IBM Event streams

Apache Kafka

This page describes connecting the Lenses Agent to Apache Kafka.

A Kafka connection is required for the agent to start. You can connect to Kafka via:

Plaintext (no credentials an unencrypted)
SSL (no credentials an encrypted)
SASL Plaintext and SASL SSL

Plaintext

With PLAINTEXT, there's no encryption and no authentication when connecting to Kafka.

The only required fields are:

kafkaBootstrapServers - a list of bootstrap servers (brokers). It is recommended to add as many brokers (if available) as convenient to this list for fault tolerance.
protocol - depending on the protocol, other fields might be necessary (see examples for other protocols)

In following example JMX metrics for Kafka Brokers are configured too, assuming that all brokers expose their JMX metrics using the same port (9581), without SSL and authentication.

SSL

With SSL the connection to Kafka is encrypted. You can also uses SSL and certificates to authenticate users against Kafka.

A truststore (with password) might need to be set explicitly if the global truststore of the Agent does not include the Certificate Authority (CA) of the brokers.

If TLS is used for authentication to the brokers in addition to encryption-in-transit, a key store (with passwords) is required.

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SSL://your.kafka.broker.0:9092
        - SSL://your.kafka.broker.1:9092
    protocol: 
      value: SSL
    sslTruststore:
      file: /path/to/truststore.jks
    sslTruststorePassword: 
      value: truststorePassword
    sslKeystore:
      file: /path/to/keystore.jks
    sslKeyPassword: 
      value: keyPassword
    sslKeystorePassword: 
      value: keystorePassword

SASL Plaintext vs SASL SSL

There are 2 SASL-based protocols to access Kafka Brokers: SASL_SSL and SASL_PLAINTEXT. They both require SASL mechanism and JAAS Configuration values. What is different is:

The transport layer is encyrpted (SSL)
The SASL mechanism for authentication (PLAIN, AWS_MSK_IAM, GSSAPI).

In addition to this, there might be a keytab file required, depending on the SASL mechanism (for example when using GSSAPI mechanism, most often used for Kerberos).

To use Kerberos authentication, a Kerberos _Connection_ should be created beforehand.

When encryption-in-transit is used (with SASL_SSL), a trust store might need to be set explicitly if the global trust store of Lenses does not include the CA of the brokers.

SASL SSL

Mechanism PLAIN

Encrypted communication and basic username and password for authentication.

provisioning.yaml

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_SSL://your.kafka.broker.0:9092
        - SASL_SSL://your.kafka.broker.1:9092
    protocol: 
      value: SASL_SSL
    sslTruststore:
      file: /path/to/truststore.jks
    sslTruststorePassword: 
      value: truststorePassword
    sslKeystore:
      file: /path/to/keystore.jks
    sslKeyPassword: 
      value: keyPassword
    sslKeystorePassword: 
      value: keystorePassword
    saslMechanism: 
      value: PLAIN
    saslJaasConfig:
      value: |
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="your-username"
        password="your-password";

Mechanism GSSAPI

In order to use Kerberos authentication, a Kerberos Connection should be created beforehand.

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_SSL://your.kafka.broker.0:9092
        - SASL_SSL://your.kafka.broker.1:9092
    protocol: 
      value: SASL_SSL
    sslTruststore:
      file: /path/to/truststore.jks
    sslTruststorePassword: 
      value: truststorePassword
    sslKeystore:
      file: /path/to/keystore.jks
    sslKeyPassword: 
      value: keyPassword
    sslKeystorePassword: 
      value: keystorePassword  
    saslMechanism: 
      value: GSSAPI
    saslJaasConfig:
      value: |
        com.sun.security.auth.module.Krb5LoginModule required
        useKeyTab=true
        storeKey=true
        useTicketCache=false
        serviceName=kafka
        principal="my-principal@DOMAIN.COM";      
     keytab:
       file: /path/to/keytab.jks

SASL Plaintext

No SSL encrypted of communication, credentials communicated to Kafka in clear text.

Mechanism SCRAM-SHA-256

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_PLAINTEXT://your.kafka.broker.0:9092
        - SASL_PLAINTEXT://your.kafka.broker.1:9092
    protocol: 
      value: SASL_PLAINTEXT
    saslMechanism: 
      value: SCRAM-SHA-256
    saslJaasConfig: 
      value: |
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="your-username"
        password="your-password";

Mechanism SCRAM-SHA-512

kafka:
- name: Kafka
  version: 1
  tags: [my-tag]
  configuration:
    kafkaBootstrapServers:
      value:
        - SASL_PLAINTEXT://your.kafka.broker.0:9092
        - SASL_PLAINTEXT://your.kafka.broker.1:9092
    protocol: 
      value: SASL_PLAINTEXT
    saslMechanism: 
      value: SCRAM-SHA-512
    saslJaasConfig: 
      value: |
        org.apache.kafka.common.security.scram.ScramLoginModule required
        username="your-username"
        password="your-password";

AWS MSK Serverless

This page describes how to connect Lenses to an Amazon MSK Serverless cluster.

It is recommended to install the Agent on an EC2 instance or with EKS in the same VPC as your MSK Serverless cluster.

Security Groups

IAM Policy

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:Connect",
                "kafka-cluster:AlterCluster",
                "kafka-cluster:DescribeCluster"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:DescribeTopic",
                "kafka-cluster:CreateTopic",
                "kafka-cluster:WriteData",
                "kafka-cluster:ReadData"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:topic/[cluster_name]/[cluster_uuid]/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "kafka-cluster:AlterGroup",
                "kafka-cluster:DescribeGroup"
            ],
            "Resource": "arn:aws:kafka:[region]:[aws_account_id]:group/[cluster_name]/[cluster_uuid]/*"
        }
    ]
}

MSK Serverless IAM to be used after cluster creation. Update this IAM policy with the relevant ARN.

Select your MSK endpoint

Click your MSK Serverless Cluster in the MSK console and select View Client Information page to check the bootstrap server endpoint.

Configure Provisioning

provisioning.yaml

kafka:
  tags: ["optional-tag"]
  name: Kafka
  configuration:
    kafkaBootstrapServers:
      value:
       - SASL_SSL://your.kafka.broker.0:9098
       - SASL_SSL://your.kafka.broker.1:9098
    protocol: SASL_SSL
    saslMechanism: 
      value: AWS_MSK_IAM
    saslJaasConfig:
      value: software.amazon.msk.auth.iam.IAMLoginModule required;
    additionalProperties:
      value:
        sasl.client.callback.handler.class: "software.amazon.msk.auth.iam.IAMClientCallbackHandle

SQL Processor Configurations

To enable the creation of SQL Processors that create consumer groups, you need to add the following statement in your IAM policy:

{
  "Action": [
    "kafka-cluster:*Topic*",
    "kafka-cluster:WriteData",
    "kafka-cluster:ReadData"
  ],
  "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
}

Update the placeholders in the IAM policy based on the relevant MSK Serverless cluster ARN.

To integrate with the AWS Glue Schema Registry, you also need to add the following statement for the registries and schemas in your IAM policy:

{
  "Action": [
    "kafka-cluster:*Group*"
  ],
  "Resource": "arn:aws:kafka:[region]:[aws_account_id]:cluster/[cluster_name]/[cluster_uuid]/*"
}

Update the placeholders in the IAM policy based on the relevant MSK Serverless cluster ARN.

To integrate with the AWS Glue Schema Registry, you also need to modify the security policy for the registry and schemas, which results in additional functions within it:

{
  "Action": [
    "glue:DeregisterDataPreview",
    "glue:ListRegistries",
    "glue:CreateRegistry",
    "glue:RegisterSchemaVersion",
    "glue:GetRegistry",
    "glue:UpdateRegistry",
    "glue:ListSchemas",
    "glue:DeleteRegistry",
    "glue:GetSchema",
    "glue:CreateSchema",
    "glue:ListSchemaVersions",
    "glue:GetSchemaVersion",
    "glue:UpdateSchema",
    "glue:DeleteSchemaVersions"
  ],
  "Resource": [
    "arn:aws:glue:[region]:[aws_account_id]:registry/*",
    "arn:aws:glue:[region]:[aws_account_id]:schema/*"
  ]
}

More details about how IAM works with MSK Serverless can be found in the documentation: MSK Serverless

Limitations

When using the Agent with MSK Serverless:

The agent does not receive Prometheus-compatible metrics from the brokers because they are not exported outside of CloudWatch.
The agent does not configure quotas and ACLs because MSK Serveless does not allow this.