Set Up a Kafka Cluster in AWS MSK

To set up a Kafka cluster in AWS MSK, perform the following steps:

1. Create a Kafka cluster in AWS MSK with Provisioned cluster type.

   

2. When configuring your AWS MSK cluster, select Availability Zones (AZs) that overlap with the AZs used by SingleStore Helios deployment.

   

3. Attach a security group that allows traffic from the internal IP range of the NLB that you will create later. You can allow traffic from all internal IP ranges.

4. During cluster creation, enable SASL/SCRAM authentication for secure authentication.

   

5. After the cluster is created:

   1. Add your SASL username and password as a secret in AWS Secrets Manager.

   2. Encrypt the secret using a Customer Managed Key (CMK).

   3. Attach the secret to the AWS MSK cluster to enable authentication for SingleStore.

      

Configure an Endpoint Service using a Single NLB

To configure an endpoint service using a Single NLB to manage connectivity to the Kafka brokers, perform the following steps:

1. Create target Groups for each Kafka broker.

   Note

   Create one target group per broker.

   1. Get the SASL/SCRAM endpoints from the Kafka cluster summary in View Client Information.

      

   2. Retrieve the IP addresses of the Kafka brokers by selecting the gear icon in the AWS MSK console and enabling the Show IP addresses option.

      

2. Create a Network Load Balancer.

   1. Create a single NLB to handle traffic for all Kafka brokers using separate listener ports.

      Note

      Only a single NLB is required for this private endpoint service setup, regardless of the number of brokers.

      

   2. Create a separate listener for each target group. Assign a unique port to each listener. For example, use ports 6001, 6002, and 6003 for the three Kafka brokers.

      

   3. Create an endpoint service pointing to the load balancer as shown in the previous step.

      

   4. After creating the endpoint service, add the AWS account ID provided by SingleStore to the Allow principals list. This enables SingleStore to find and access the private endpoint service.

      

   5. Use the endpoint service created above to set up an outbound endpoint to SingleStore, which allows SingleStore to connect to your Kafka service. For more information, refer to Set up an Outbound Connection in SingleStore.

Set Up an Outbound Connection in SingleStore

Refer to Connect to SingleStore Helios using AWS PrivateLink for more information on configuring the outbound private connection in SingleStore Helios.

Create a Pipeline in SingleStore

Using the outbound private connection created earlier, you can create a pipeline in SingleStore. For example,

CREATE OR REPLACE PIPELINE KAFKA_MUNIS_POC_PIPELINE AS LOAD DATA KAFKA 'b-2.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096,b-3.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096,b-1.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096/test'
CONFIG '{
"spoof.dns": {
    "b-1.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"<VPC ENDPOINT>:<NLB LISTENER PORT1>",
    "b-2.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"<VPC ENDPOINT>:<NLB LISTENER PORT2>",
    "b-3.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"<VPC ENDPOINT>:<NLB LISTENER PORT3>"
  },
  "sasl.username": "<REDACTED>",
  "sasl.mechanism": "SCRAM-SHA-512",
  "security.protocol": "SASL_SSL",
  "ssl.ca.location": "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem"}'
  CREDENTIALS '{"sasl.password": "<REDACTED>"}'
  DISABLE OUT_OF_ORDER OPTIMIZATION
  SKIP DUPLICATE KEY ERRORS
  INTO TABLE kafkatest;

Ensure the following configuration details are applied:

1. In the LOAD DATA section, use the appropriate Kafka port for your connection method, e.g., 9096 in this case for SASL/SCRAM.

2. In the spoof.dns section, define mappings that connect each Kafka broker to the corresponding outbound PrivateLink endpoint.

3. Each broker’s hostname is mapped to the correct NLB listener port (e.g., 6001, 6002, 6003).

CREATE OR REPLACE PIPELINE KAFKA_MUNIS_POC_PIPELINE AS LOAD DATA KAFKA 'b-2.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096,b-3.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096,b-1.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096/test'
CONFIG '{
"spoof.dns": {
    "b-1.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"vpce-0af45e49ab211df13-67n9i1aw-us-east-1c.vpce-svc-02e49c31898768e9c.us-east-1.vpce.amazonaws.com:6003",
    "b-2.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"vpce-0af45e49ab211df13-67n9i1aw-us-east-1a.vpce-svc-02e49c31898768e9c.us-east-1.vpce.amazonaws.com:6001",
    "b-3.testvibhor3.c7wkh8.c10.kafka.us-east-1.amazonaws.com:9096":"vpce-0af45e49ab211df13-67n9i1aw-us-east-1b.vpce-svc-02e49c31898768e9c.us-east-1.vpce.amazonaws.com:6002"
  },
  "sasl.username": "<REDACTED>",
  "sasl.mechanism": "SCRAM-SHA-512",
  "security.protocol": "SASL_SSL",
  "ssl.ca.location": "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem"}'
  CREDENTIALS '{"sasl.password": "<REDACTED>"}'
  DISABLE OUT_OF_ORDER OPTIMIZATION
  SKIP DUPLICATE KEY ERRORS
  INTO TABLE kafkatest;