Load Data from Kafka
On this page
Securely Connect to Kafka from SingleStore
Overview
When running a CREATE PIPELINE .
statement, you may need to make a secure connection to Kafka.
Use Secure Socket Layer (SSL) for the connection and Simple Authentication and Security Layer (SASL) to authenticate.
-
GSSAPI (Kerberos)
-
PLAIN
-
SCRAM-SHA-256
-
SCRAM-SHA-512
-
OAUTHBEARER SASL
Kerberos may used for authentication.
This topic assumes SSL and/or Kerberos have been set up, configured, and enabled on the Kafka brokers.
Convert Java Keystore (JKS) to Privacy Enhanced Mail (PEM) Key
To use SSL encryption for SingleStore pipelines, JKS keys need to be converted to PEM keys.
Note
In the steps below, the < > symbols indicate a variable and any information provided between these symbols is an example.
-
Create the key and keystore.
You will be prompted to enter details such as name, organizational unit, city, state, etc. keytool -genkey -keyalg RSA -keystore <keystore-name>.jks -storepass <password> -alias "<keystore-alias-name>"What is your first and last name? [Unknown]:- What is the name of your organizational unit? [Unknown]:- What is the name of your organization? [Unknown]:-
-
Export the client certificate from the keystore using the same password as in step 1.
keytool -exportcert -rfc -file <pem-name>.pem -alias <pem-name-alias> -keystore <keystore-name>.jksEnter keystore password: <password> Certificate stored in file <pem-name>.pem
-
Import the client certificate to the truststore located on your Apache Server.
Enter a new password for the keystore. keytool -keystore <truststore-name>.jks -alias <truststore-name-alias> -import -file <pem-name>.pemEnter keystore password: Re-enter new password: Trust this certificate? [no]: yes Certificate was added to keystore
-
Convert the client keystore to Public-Key Cryptography Standards (PKCS12) format.
The <key name>. jks and password are the same as in step 1. keytool -v -importkeystore -srckeystore <keystore-name>.jks -srcalias <keystore alias> -destkeystore <new-keystore-name>.p12 -deststoretype PKCS12Importing keystore <keyname>.jks to <new-keyname>.p12... Enter destination keystore password: Re-enter new password: Enter source keystore password: [Storing new-keystore-name.p12]
-
Extract the client certificate key into a .
pem file. Use the import password created in step 4. openssl pkcs12 -in <new-keyname>.p12 -nocerts -nodes > <keyname>.pemEnter Import Password:
Note
Refer to Generating SSL Certificates for more information.
Steps for Creating a Secure Connection
To create a secure connection from SingleStore to a Kafka cluster, follow these steps in order.
Copy Security Files to the SingleStoreCluster
Securely copy the CA certificate, SSL certificate, and SSL key used for connections between the SingleStorecluster and Kafka brokers from the Kafka cluster to each SingleStore node.scp
, to copy the files to your SingleStore nodes.
Setup Your SingleStoreCluster for Kerberos Authentication
To configure a Kafka pipeline to authenticate with Kerberos, you must configure all of your nodes in your SingleStorecluster as clients for Kerberos authentication.
-
Securely copy the keytab file containing the SingleStore service principal (e.
g. memsql/host.
) from the Kerberos server to each node in your SingleStorecluster.domain. com@REALM. NAME You should use a secure file transfer method, such as scp
, to copy the keytab file to your SingleStore nodes.The file location on your SingleStore nodes should be consistent across the cluster. -
Make sure your SingleStore nodes can connect to the KDC server using the fully-qualified domain name (FQDN) of the KDC server.
This might require configuring network settings or updating /etc/hosts
on your nodes. -
Also ensure that the
memsql
service account on each node can access the copied keytab file.This can be accomplished by changing file ownership or permissions. If the memsql
account cannot access the keytab file, you will not be able to complete the next step because your master aggregator will not be able to restart after applying configuration updates. -
When authenticating with Kerberos, SingleStore needs to authenticate as a client, which means you must also install a Kerberos client onto each node in your cluster.
The following installs the krb5-user
package for Debian-based Linux distributions.sudo apt-get update && apt-get install krb5-userWhen setting up your Kerberos configuration settings, set your default realm, Kerberos admin server, and other options to those defined by your KDC server.
In the examples used in this topic, the default realm is EXAMPLE.
, and the Kerberos server settings are set to the FQDN of the KDC serverCOM host.
.example. com
Build a String with the Connection Settings
Using the following settings, create a string containing the CONFIG
clause and optionally the CREDENTIALS
clause of the CREATE PIPELINE .
or SELECT .
statement that you will be running.
SSL Connection Settings
-
In your
CONFIG
JSON, if you want to enable SSL encryption only, set"security.
.protocol": "ssl" If you want to enable Kerberos with SSL, or otherwise want to use SASL, set "security.
.protocol": "sasl_ ssl" -
Set the remaining SSL configuration in the
CONFIG
JSON:-
ssl.
: Path to the CA certificate on the SingleStore node.ca. location -
ssl.
: Path to the SSL certificate on the SingleStore node.certificate. location -
ssl.
: Path to the SSL certificate key on the SingleStore node.key. location
-
-
If your SSL certificate key is using a password, set it in your
CREDENTIALS
JSON.-
ssl.
: Password for the SSL certificate key.key. password
-
SASL Connection Settings
-
In your
CONFIG
JSON, set"security.
for SSL connections, orprotocol": "sasl_ ssl" "security.
if you want to authenticate with Kafka without SSL encryption.protocol": "sasl_ plaintext" -
If your Kafka brokers do not use SCRAM for authentication, set
"sasl.
in yourmechanism": "PLAIN" CONFIG
JSON.Otherwise, set "sasl.
ormechanism": "SCRAM-SHA-256" "sasl.
.mechanism": "SCRAM-SHA-512" -
In your
CONFIG
JSON, provide the username,"sasl.
.username": "<kafka_ credential_ username>" -
In your
CREDENTIALS
JSON, provide the password,"sasl.
.password": "<kafka_ credential_ password>"
Note
SASL_
authentication mode with Kafka sends your credentials unencrypted over the network.
SASL_
authentication mode with Kafka will encrypt the credentials information sent over the network, but transport of Kafka messages themselves is not secure.
Kerberos Connection Settings
-
In your
CONFIG
JSON, set"sasl.
.mechanism": "GSSAPI" -
Set
"security.
for Kerberos and SSL connections, orprotocol": "sasl_ ssl" "security.
if you want to authenticate with Kerberos without SSL encryption.protocol": "sasl_ plaintext" -
Set the remaining Kerberos configuration in
CONFIG
JSON:-
sasl.
: The Kerberos principal name that Kafka runs as.kerberos. service. name For example, "kafka"
. -
sasl.
: The local file path on the SingleStore node to the authenticating keytab.kerberos. keytab -
sasl.
: The service principal name for the SingleStorecluster.kerberos. principal For example, "memsql/host.
.example. com@EXAMPLE. COM"
-
Configuring OAUTHBEARER Authentication Mechanism
OAUTHBEARER is used to secure access to resources on a server by requiring clients to obtain a bearer token.
To use SASL OAUTHBEARER authentication with Kafka, the following information is required:
-
"sasl.
- Specifies the client will authenticate using an OAuth 2.mechanism":"OAUTHBEARER" 0 Bearer Token. -
"sasl.
- The client ID is usually provided by the OAuth provider when the client is registered.oauthbearer. client. id":"<CLIENT_ ID>" It is a unique ID that is associated with the OAuth 2. 0 Bearer Token. -
"sasl.
The client secret is usually assigned by the OAuth provider when a client is registered and is used to authenticate the client.oauthbearer. client. secret":"<CLIENT_ SECRET>" -
"sasl.
- This is the endpoint URL on the authorization server that is used to obtain an OAuth 2.oauthbearer. token. endpoint. url":"<ENDPOINT_ URL>" 0 Bearer Token. The client sends a request to this endpoint to get a token, which is then used to authenticate subsequent requests to the server.
Optional configurations are:
-
"sasl.
- Determines the permissions and resources that are available to an authorized client.oauthbearer. scope":"<SCOPE>" This extension is optional. -
"sasl.
- Can be included in the SASL/OAUTHBEARER mechanism to provide additional data or parameters for authentication.outhbearer. extensions":"<EXTENSION>" Consult RFC-7628 for further information on SASL extensions. -
See "OAUTHBEARER Pipelines Configuration Details" section for additional details regarding "sasl.
oauthbearer. ssl. ca. location" "sasl. oauthbearer. config" fields.
Prerequisites
To use SASL OAUTHBEARER authentication the following prerequisites are required:
-
A Kafka broker with listeners configured using OAUTHBEARER authentication.
-
A connection to a database where Kafka brokers are reachable to create a pipeline to pull data from the Kafka queue.
-
An identity service was selected (e.
g. , Okta, Google OAuth, Facebook OAuth, Azure AD, Keycloak, etc. ). -
An Oauthbearer client was created and configured with the client_
credentials grant type on the identity service.
Note
The instructions for setting up a client on the Identity Service will vary depending on the server chosen and the type of application being built.
Syntax for SASL OAUTHBEARER Pipeline
Below is the syntax for creating a Kafka pipeline using OAUTHBEARER authentication.
CREATE or REPLACE PIPELINE <pipeline_name>AS LOAD DATA KAFKA "<Kafka cluster location>"CONFIG '{"security.protocol":"SASL_SSL","sasl.mechanism":"OAUTHBEARER","sasl.oauthbearer.client.id":"<CLIENT_ID>","sasl.oauthbearer.client.secret":<"CLIENT_SECRET>","sasl.oauthbearer.token.endpoint.url":"<ENDPOINT_URL>","sasl.oauthbearer.scope":"<SCOPE>"}'INTO TABLE <table_name>;
OAUTHBEARER Pipelines Configuration Details
To review the current pipeline configuration use the command:
SELECT * FROM information_schema.pipelines
Most token providers are assumed to work out-of-the-box.
Token Request Details:
-
SingleStore implements
client_
grant type for OAUTHBEARER token requests.credentials Ensure the OAuth client is created with support for the client_
grant type.credentials -
By default, SingleStore treats Oauthbearer tokens as opaque to address privacy concerns and uses the expires_
in field from the JSON token response to determine the token's validity time. If expires_
is not present orin "sasl.
is set in the pipeline configuration, we fall back to decoding the Oauth token JWT and use the exp claim to determine its validity.oauthbearer. config”:"use_ expires_ in=false” -
The token refresh is scheduled at 80% of the token’s lifetime and handled by SingleStore in the background.
-
For https Oauth token requests, by default, SingleStore will use the CA bundle specified at pipeline creation or try to find one of the system paths if none are specified.
This behavior can be changed by including additional configuration settings in the pipeline CONFIG on creation. This change will affect token request logic only, not the total SSL Kafka communication. -
"sasl.
- to use the system default path;oauthbearer. ssl. ca. location":"system" -
"sasl.
- to use a specific CA path location;oauthbearer. ssl. ca. location":"/usr/lib/ssl/certs/ca-certificates. crt" -
"sasl.
(empty) to disable SSL verification;oauthbearer. ssl. ca. location":""
-
Token Cache Details:
-
The OAUTHBEARER tokens are cached in the extractor pools for some time (along with the extractors) and the pipeline cookies (per pipeline).
The Oauth server must allow multiple simultaneous active tokens due to the several nodes or extractor pools used in the SingleStore architecture. -
"sasl.
- by default, tokens are cached in the pipeline cookies to minimize the number of token requests.oauthbearer. cookie. cache":"on" To disable this feature set "sasl.
.oauthbearer. cookie. cache":"off" To reset the pipeline cookies, use alter pipeline … set offset cursor ''
. -
The pipeline may fail because the OAUTHBEARER client config on the identity server is changed or the pipeline config is altered in a way which causes previously requested cached tokens to become invalid.
This issue may be avoided by resetting the respective pipeline cache. -
alter pipeline … set offset cursor ''
- to reset cookies cache; -
flush extractor pools
- to reset extractor pools;
-
Kafka Version Setting
Warning
Using SSL and SASL with Kafka requires Kafka protocol version 0.CREATE PIPELINE .
and SELECT .
statements using SSL and SASL with Kafka also need to adhere to that version requirement.CONFIG
clause, similar to this CONFIG '{"kafka_
.pipelines_
engine variable controls this parameter for any pipeline without using a Kafka version configuration value in a CREATE PIPELINE .
statement.
Final Step: Use the Connection String in a SQL Statement
Create your CREATE PIPELINE .
or SELECT .
statement, using the string containing the connection settings that you created in the previous steps.
Examples
The following examples make the following assumptions:
-
Port 9092 is a plaintext endpoint
-
Port 9093 is an SSL endpoint
-
Port 9094 is a plaintext SASL endpoint
-
Port 9095 is an SSL SASL endpoint
Note
The examples use CREATE PIPELINE
, but the CONFIG
and CREDENTIALS
clauses shown can be used with SELECT .
also.SELECT .
.
Plaintext
The following CREATE PIPELINE
statements are equivalent:
CREATE PIPELINE `kafka_plaintext`AS LOAD DATA KAFKA 'host.example.com:9092/test'CONFIG '{"security.protocol": "plaintext"}'INTO table t;
CREATE PIPELINE `kafka_no_creds`AS LOAD DATA KAFKA 'host.example.com:9092/test'INTO table t;
SSL
CREATE PIPELINE `kafka_ssl`AS LOAD DATA KAFKA 'host.example.com:9093/test'CONFIG '{"security.protocol": "ssl","ssl.certificate.location": "/var/private/ssl/client_memsql_client.pem","ssl.key.location": "/var/private/ssl/client_memsql_client.key","ssl.ca.location": "/var/private/ssl/ca-cert.pem"}'CREDENTIALS '{"ssl.key.password": "abcdefgh"}'INTO table t;
Kerberos with no SSL
CREATE PIPELINE `kafka_kerberos_no_ssl`AS LOAD DATA KAFKA 'host.example.com:9094/test'CONFIG '{""security.protocol": "sasl_plaintext","sasl.mechanism": "GSSAPI","sasl.kerberos.service.name": "kafka","sasl.kerberos.principal": "memsql/host.example.com@EXAMPLE.COM","sasl.kerberos.keytab": "/etc/krb5.keytab"}'INTO table t
Kerberos with SSL
CREATE PIPELINE `kafka_kerberos_ssl`AS LOAD DATA KAFKA 'host.example.com:9095/test'CONFIG '{"security.protocol": "sasl_ssl","sasl.mechanism": "GSSAPI","ssl.certificate.location": "/var/private/ssl/client_memsql_client.pem","ssl.key.location": "/var/private/ssl/client_memsql_client.key","ssl.ca.location": "/var/private/ssl/ca-cert.pem","sasl.kerberos.service.name": "kafka","sasl.kerberos.principal": "memsql/host.example.com@EXAMPLE.COM","sasl.kerberos.keytab": "/etc/krb5.keytab"}'CREDENTIALS '{"ssl.key.password": "abcdefgh"}'INTO table t
SASL/PLAIN with SSL
CREATE PIPELINE `kafka_sasl_ssl_plain`AS LOAD DATA KAFKA 'host.example.com:9095/test'CONFIG '{"security.protocol": "sasl_ssl","sasl.mechanism": "PLAIN","ssl.certificate.location": "/var/private/ssl/client_memsql_client.pem","ssl.key.location": "/var/private/ssl/client_memsql_client.key","ssl.ca.location": "/var/private/ssl/ca-cert.pem","sasl.username": "kafka"}'CREDENTIALS '{"ssl.key.password": "abcdefgh", "sasl.password": "metamorphosis"}'INTO table t;
SASL/PLAIN without SSL
CREATE PIPELINE `kafka_sasl_plaintext_plain`AS LOAD DATA KAFKA 'host.example.com:9094/test'CONFIG '{"security.protocol": "sasl_plaintext","sasl.mechanism": "PLAIN","sasl.username": "kafka"}'CREDENTIALS '{"sasl.password": "metamorphosis"}'INTO table t;
SASL/SCRAM with SSL
CREATE PIPELINE `kafka_sasl_ssl_scram`AS LOAD DATA KAFKA 'host.example.com:9095/test'CONFIG '{"security.protocol": "sasl_ssl","sasl.mechanism": "SCRAM-SHA-512","ssl.certificate.location": "/var/private/ssl/client_memsql_client.pem","ssl.key.location": "/var/private/ssl/client_memsql_client.key","ssl.ca.location": "/var/private/ssl/ca-cert.pem","sasl.username": "kafka"}'CREDENTIALS '{"ssl.key.password": "abcdefgh", "sasl.password": "metamorphosis"}'INTO table t;
SASL/SCRAM without SSL
CREATE PIPELINE `kafka_sasl_plaintext_plain`AS LOAD DATA KAFKA 'host.example.com:9094/test'CONFIG '{"security.protocol": "sasl_plaintext","sasl.mechanism": "SCRAM-SHA-512","sasl.username": "kafka"}'CREDENTIALS '{"sasl.password": "metamorphosis"}'INTO table t;
Configure SASL OAUTHBEARER for Use with Kafka Pipelines
Note
SASL OAUTHBEARER is not supported with SELECT INTO .
CREATE or REPLACE PIPELINE <pipeline_name>AS LOAD DATA KAFKA "<Kafka cluster location>"CONFIG '{"security.protocol":"SASL_SSL","sasl.mechanism":"OAUTHBEARER","sasl.oauthbearer.client.id":"<CLIENT_ID>","sasl.oauthbearer.client.secret":<"CLIENT_SECRET>","sasl.oauthbearer.token.endpoint.url":"<ENDPOINT_URL>","sasl.oauthbearer.scope":"<SCOPE>"}'INTO TABLE <table_name>;
Load Data from the Confluent Kafka Connector
The SingleStore Confluent Kafka Connector is a Kafka Connect connector that allows you to easily ingest AVRO, JSON, and CSV messages from Kafka topics into SingleStore.
Learn more about the SingleStore Confluent Kafka Connector, including how to install and configure it, in Working with the Kafka Connector.
Connecting to Confluent Schema Registry over SSL
To enable Pipelines to connect to Confluent Schema Registry over SSL, install the registry’s certificate (ca-cert) on all nodes in your SingleStore cluster.
cat ca-cert >> /etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pemupdate-ca-trust
Note
You only need to install the registry’s certificate (ca-cert) in your SingleStore cluster if you are using a self-signed certificate or a certificate signed by an internal CA.
After installing the registry’s certificate on all nodes, existing Pipelines that you already created (or new Pipelines that you create) with the host name/IP address and port of Confluent Schema Registry will use SSL automatically.
Schema Registry Configuration Options
Supported configuration options include:
-
schema.
registry. ssl. ca. location -
schema.
registry. ssl. ca. directory -
schema.
registry. ssl. verifypeer -
schema.
registry. ssl. certificate. location -
schema.
registry. ssl. key. location -
schema.
registry. ssl. key. password -
schema.
registry. curl. timeout -
schema.
registry. curl. tcp. keepalive -
schema.
registry. curl. tcp. keepidle -
schema.
registry. curl. keepintvl -
schema.
registry. curl. connecttimeout -
schema.
registry. username -
schema.
registry. password -
schema.
registry. curl. verbose
In this section
Last modified: July 24, 2025