How to Set Up Redshift Streaming Ingestion with SSL-Enabled Kafka on EC2

11 minute read

Content level: Intermediate

This post explain the step-by-step to configure Redshift Streaming Ingestion with self-managed Kafka on EC2 using TLS/SSL listener

In this post, we will demonstrate how Amazon Redshift Streaming Ingestion can consume Kafka topics from a self-managed Kafka cluster running on Amazon EC2. This setup can also be used to connect to any Apache Kafka installation in the cloud or on-premises environment. Amazon Redshift requires TLS transport when establishing connections to Apache Kafka, supporting either mTLS authentication or no authentication. For TLS transport, Amazon Redshift only supports certificates generated by publicly trusted certificate authorities.

Below is the architecture diagram depicting the complete environment setup for Redshift Streaming Ingestion with self-managed Kafka on EC2:

Enter image description here

Prerequisites

To set up this demo, we need the following prerequisites:

A VPC with private and public subnets that have sufficient available IP addresses.
A NAT Gateway for outbound internet access from private subnets.
Sufficient IAM user/role permissions to execute AWS commands.

Set Up Kafka on EC2

Disclaimer: This guide demonstrates a single-broker Kafka setup for testing and demonstration purposes only. It does not cover the steps to create a redundant Kafka cluster with multiple brokers. For production environments, you should follow Kafka best practice guidelines to set up a production-grade, self-managed Kafka cluster

We will use an EC2 instance with Amazon Linux 2023 to run Kafka using Docker. Launch the EC2 instance in the private subnet of your VPC. Here is the EC2 user data script that will install Docker, Docker Compose, and Java Keytool:

#!/bin/bash
dnf update -y
dnf install -y docker java-17-amazon-corretto-devel
systemctl start docker
systemctl enable docker
usermod -a -G docker ec2-user

# Install docker-compose
curl -L "https://github.com/docker/compose/releases/download/v2.24.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
chmod +x /usr/local/bin/docker-compose

# Create folder for Docker Kafka setup later
mkdir -p /home/ec2-user/kafka-docker/ssl      
chown -R ec2-user:ec2-user /home/ec2-user/kafka-docker

Important: Note down the private IP address of the EC2 instance after launch.

# Get private IP address
TOKEN=$(curl -s -X PUT "http://169.254.169.254/latest/api/token" -H "X-aws-ec2-metadata-token-ttl-seconds: 21600")
PRIVATE_IP=$(curl -s -H "X-aws-ec2-metadata-token: $TOKEN" -s http://169.254.169.254/latest/meta-data/local-ipv4)
echo $PRIVATE_IP

Configure the EC2 security group to allow inbound traffic on ports 9092 and 9093 for Kafka access.

Prepare the public SSL certificate. You will need three files from your certificate authority:

certificate.txt (your domain certificate)
certificate_chain.txt (intermediate certificates)
private_key.txt (your private key)

Convert these files to the required Java KeyStore format using openssl and keytool:

# Set password
export PASSWORD=<YOUR PASSWORD>

# Convert your public certificate to PKCS12 format
openssl pkcs12 -export -in certificate.txt -inkey private_key.txt -out kafka.p12 -name kafka -certfile certificate_chain.txt -password pass:$PASSWORD

# Convert to JKS
keytool -importkeystore -deststorepass $PASSWORD -destkeypass $PASSWORD -destkeystore kafka.server.keystore.jks -srckeystore kafka.p12 -srcstoretype PKCS12 -srcstorepass $PASSWORD -alias kafka

# Create Keystore credential
echo $PASSWORD > keystore_creds
echo $PASSWORD > key_creds

Note: If you use AWS Certificate Manager, you can now create exportable public SSL certificate. See this blog for the detail step to export a public certificate.

Copy all four generated files (kafka.p12, kafka.server.keystore.jks, key_creds, keystore_creds) to EC2 folder /home/ec2-user/kafka-docker/ssl

Create a docker-compose.yml file in /home/ec2-user/kafka-docker/ using the following template:

services:
kafka:
    image: apache/kafka-native:3.8.1
    ports:
    - "9092:9092"
    - "9093:9093"
    volumes:
    - ./ssl:/etc/kafka/secrets
    environment:
    KAFKA_NODE_ID: 1
    KAFKA_PROCESS_ROLES: broker,controller
    KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093,CONTROLLER://localhost:9094
    KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://<PUBLIC_DNS_RECORD>:9092,SSL://<PUBLIC_DNS_RECORD>:9093
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT
    KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
    KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9094
    KAFKA_SSL_KEYSTORE_FILENAME: kafka.server.keystore.jks
    KAFKA_SSL_KEYSTORE_CREDENTIALS: keystore_creds
    KAFKA_SSL_KEY_CREDENTIALS: key_creds
    KAFKA_SSL_CLIENT_AUTH: none
    KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
    KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
    KAFKA_MESSAGE_MAX_BYTES: 1000000000
    KAFKA_REPLICA_FETCH_MAX_BYTES: 1000000000
    KAFKA_SOCKET_REQUEST_MAX_BYTES: 1000000000

Important: Replace both <PUBLIC_DNS_RECORD> entries in KAFKA_ADVERTISED_LISTENERS with the DNS record associated with your certificate. This configuration creates a Kafka container with both plaintext (port 9092) and SSL (port 9093) listeners.

Deploy the Docker container using docker-compose up. Use docker-compose up -d to run in the background without logging output.

cd /home/ec2-user/kafka-docker/
docker-compose up

Verify that Docker is listening on the correct ports using netstat command as shown below:

netstat -an | egrep "9092|9093"

DNS Configuration for Redshift Connectivity: Since Amazon Redshift Streaming ingestion requires public certificates, Redshift must connect to Kafka using DNS Record instead of IP address. You have two options:
- Option 1: Public DNS Record: If you have access public DNS hosted zone associated with your certificate, create an A record pointing to the EC2 private IP address.
- Option 2: Split-View DNS (Recommended for security)
  - Create a Route53 Private Hosted Zone with the same domain name as your certificate.
  - Associate the Private Hosted Zone with the Redshift's VPC.
  - In this Private Hosted Zone, create A record pointing to the EC2 private IP address.
Example configuration:
- Public domain: example.com (managed as Route53 Public Hosted Zone)
- SSL certificate: demokafka.example.com (using AWS Certificate Manager)
- Private Hosted Zone: example.com (attached to VPC where Redshift created)
- A record in Private Hosted Zone: demokafka.example.com -> EC2 private IP
- Important: Ensure that the VPC has both DnsHostnames and DnsSupport enabled.

Set Up Kafka Client EC2

To test the Kafka installation, create a secondary EC2 instance in the same VPC.

Launch an Amazon Linux 2023 EC2 instance (t3.large) in the private subnet and use the following user data script:

#!/bin/bash
dnf update -y
dnf -y install java-17-amazon-corretto python3 python3-pip
cd /home/ec2-user
wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgz
tar xzvf *tgz
chown -R ec2-user:ec2-user kafka*

cd kafka*/libs/ && wget https://github.com/aws/aws-msk-iam-auth/releases/download/v2.3.0/aws-msk-iam-auth-2.3.0-all.jar

# Create client.properties file
cat << EOF > ../config/client.properties
security.protocol=PLAINTEXT
request.timeout.ms=30000
admin.request.timeout.ms=30000
default.api.timeout.ms=30000
metadata.max.age.ms=5000
connections.max.idle.ms=30000
EOF

cat << EOF > ../config/client-ssl.properties
security.protocol=SSL
request.timeout.ms=30000
admin.request.timeout.ms=30000
default.api.timeout.ms=30000
metadata.max.age.ms=5000
connections.max.idle.ms=30000
EOF

# Create requirements.txt
cat << EOF > /home/ec2-user/requirements.txt
kafka-python==2.0.2
EOF

# Create telco_data_generator.py
cat << 'PYTHON_EOF' > /home/ec2-user/telco_data_generator.py
#!/usr/bin/env python3

import json
import random
import time
from datetime import datetime, timedelta
from kafka import KafkaProducer

class TelcoDataGenerator:
    def __init__(self, bootstrap_servers='localhost:9092', use_ssl=False):
        producer_config = {
            'bootstrap_servers': bootstrap_servers,
            'value_serializer': lambda v: json.dumps(v).encode('utf-8')
        }

        if use_ssl:
            producer_config.update({
                'security_protocol': 'SSL',
                'ssl_check_hostname': False,
                'ssl_cafile': None
            })

        self.producer = KafkaProducer(**producer_config)

        # Sample subscriber IDs
        self.subscriber_ids = [f"62812{random.randint(10000000, 99999999)}" for _ in range(1000)]

        # Service types and rates
        self.services = {
            'voice': {'rate_per_minute': 500, 'unit': 'minutes'},
            'sms': {'rate_per_unit': 150, 'unit': 'messages'},
            'data': {'rate_per_mb': 50, 'unit': 'MB'},
            'roaming_voice': {'rate_per_minute': 2000, 'unit': 'minutes'},
            'roaming_data': {'rate_per_mb': 500, 'unit': 'MB'}
        }

    def generate_usage_event(self):
        subscriber_id = random.choice(self.subscriber_ids)
        service_type = random.choice(list(self.services.keys()))
        service_info = self.services[service_type]

        # Generate usage amount based on service type
        if service_type == 'voice' or service_type == 'roaming_voice':
            usage_amount = round(random.uniform(0.5, 30.0), 2)  # minutes
        elif service_type == 'sms':
            usage_amount = random.randint(1, 10)  # messages
        elif service_type == 'data':
            usage_amount = round(random.uniform(1.0, 500.0), 2)  # MB
        elif service_type == 'roaming_data':
            usage_amount = round(random.uniform(0.1, 50.0), 2)  # MB

        # Calculate cost
        if 'rate_per_unit' in service_info:
            rate = service_info['rate_per_unit']
        elif 'rate_per_minute' in service_info:
            rate = service_info['rate_per_minute']
        else:
            rate = service_info['rate_per_mb']

        cost = usage_amount * rate

        event = {
            'event_id': f"evt_{int(time.time() * 1000)}_{random.randint(1000, 9999)}",
            'subscriber_id': subscriber_id,
            'timestamp': datetime.now().isoformat(),
            'service_type': service_type,
            'usage_amount': usage_amount,
            'unit': service_info['unit'],
            'cost_idr': int(cost),
            'location': random.choice(['Jakarta', 'Surabaya', 'Bandung', 'Medan', 'Makassar']),
            'cell_id': f"CELL_{random.randint(1000, 9999)}",
            'session_id': f"sess_{random.randint(100000, 999999)}"
        }

        return event

    def start_generating(self, topic='telco-usage', events_per_second=10, duration_seconds=None):
        print(f"Starting telco data generator...")
        print(f"Topic: {topic}")
        print(f"Events per second: {events_per_second}")

        start_time = time.time()
        event_count = 0

        try:
            while True:
                if duration_seconds and (time.time() - start_time) > duration_seconds:
                    break

                event = self.generate_usage_event()
                self.producer.send(topic, value=event)
                event_count += 1

                if event_count % 100 == 0:
                    print(f"Generated {event_count} events")

                time.sleep(1.0 / events_per_second)

        except KeyboardInterrupt:
            print("\nStopping generator...")
        finally:
            self.producer.flush()
            self.producer.close()
            print(f"Total events generated: {event_count}")

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description='Telco Usage Data Generator')
    parser.add_argument('--bootstrap-servers', default='localhost:9092', help='Kafka bootstrap servers')
    parser.add_argument('--topic', default='telco-usage', help='Kafka topic name')
    parser.add_argument('--rate', type=int, default=10, help='Events per second')
    parser.add_argument('--duration', type=int, help='Duration in seconds (optional)')
    parser.add_argument('--ssl', action='store_true', help='Use SSL/TLS connection')

    args = parser.parse_args()

    generator = TelcoDataGenerator(args.bootstrap_servers, args.ssl)
    generator.start_generating(args.topic, args.rate, args.duration)
PYTHON_EOF

# Install Python dependencies and set permissions
pip3 install -r /home/ec2-user/requirements.txt
chmod +x /home/ec2-user/telco_data_generator.py
chown -R ec2-user:ec2-user /home/ec2-user/
chmod 600 /home/ec2-user/kafka*/config/client*.properties

This user data script performs the following tasks:

Install Kafka client tools testing
Sets up Python and required packages
Create a Python data generator that simulates telecommunications usage data (/home/ec2-user/telco_data_generator.py).

The script will generates telco events in JSON format as shown in the following example:

{"event_id": "evt_1759737265754_7991", "subscriber_id": "6281280381900", "timestamp": "2025-10-06T07:54:25.754659", "service_type": "roaming_voice", "usage_amount": 1.91, "unit": "minutes", "cost_idr": 3820, "location": "Bandung", "cell_id": "CELL_6155", "session_id": "sess_136265"}

{"event_id": "evt_1759737265795_8001", "subscriber_id": "6281226874491", "timestamp": "2025-10-06T07:54:25.795231", "service_type": "data", "usage_amount": 465.02, "unit": "MB", "cost_idr": 23251, "location": "Jakarta", "cell_id": "CELL_4282", "session_id": "sess_653376"}

Log into client EC2 instance and test connectivity to the Kafka EC2:

curl -v <KAFKA_EC2_IP> 9092
curl -v <KAFKA_EC2_IP> 9093

curl -v <DNS_RECORD> 9092
curl -v <DNS_RECORD> 9093

Expected output for successful connection:

[ec2-user@ip-10-0-16-54 ~]$ curl -v  10.0.26.30:9092
*   Trying 10.0.26.30:9092...
* Connected to 10.0.26.30 (10.0.26.30) port 9092

Create a test topic using SSL connection:

cd ~/kafka*/
bin/kafka-topics.sh --create --topic redshiftdemo --bootstrap-server <DNS_RECORD>:9093 --partitions 1 --replication-factor 1 --command-config config/client-ssl.properties

Note: in the sample I use redshiftdemo as topic name.

Generate sample data to the topic using the Python script:

python3 ~/telco_data_generator.py  --bootstrap-servers <DNS_RECORD>:9093 --topic redshiftdemo --rate 10 --duration 1 --ssl

Verify data is being produced by consuming the topic:

cd ~/kafka_2.13-3.6.0 && bin/kafka-console-consumer.sh --bootstrap-server <DNS_RECORD>:9093  --topic redshiftdemo --from-beginning --consumer.config config/client-ssl.properties

Set Up Redshift Streaming Ingestion

You can use either a Redshift provisioned cluster or Redshift Serverless to ingest data directly from Kafka on EC2. Ensure that:

For Redshift Serverless, you have enabled Enhanced VPC routing
You have network connectivity between the Redshift security group to the EC2 security group

To configure Streaming Ingestion, follow these steps:

Open the Redshift Query Editor and connect to your database.

Create an external schema using the following command:

-- Create external schema for EC2
CREATE EXTERNAL SCHEMA demo_kafka_onprem
FROM KAFKA
AUTHENTICATION none
URI '<PUBLIC_DNS_RECORD>:9093';

Create a materialized view associated with the Kafka topic you created earlier:

-- Create MV from Kafka on EC2
CREATE MATERIALIZED VIEW demo_kafka_mv AUTO REFRESH YES AS
SELECT
    kafka_partition,
    kafka_offset,
    refresh_time,
    JSON_PARSE(kafka_value) as Data
FROM
    demo_kafka_onprem."redshiftdemo";

Note: The materialized view will automatically refresh and ingest new data from the Kafka topic in real-time.

Query the materialized view to verify data ingestion:

SELECT 
    data.event_id::VARCHAR(60) AS event_id,
    data.subscriber_id::VARCHAR(60) AS subscriber_id,
    data.timestamp::TIMESTAMP AS timestamp,
    data.service_type::VARCHAR(20) AS service_type,
    data.usage_amount::DECIMAL(10,2) AS usage_amount,
    data.unit::VARCHAR(20) AS unit,
    data.cost_idr::INTEGER AS cost_idr,
    data.location::VARCHAR(50) AS location,
    data.cell_id::VARCHAR(20) AS cell_id,
    data.session_id::VARCHAR(20) AS session_id
FROM 
    demo_kafka_mv;

(Optional) Create a parsed view for easier querying:

CREATE  VIEW telco_data_parsed_onprem AS
SELECT 
    data.event_id::VARCHAR(60) AS event_id,
    data.subscriber_id::VARCHAR(60) AS subscriber_id,
    data.timestamp::TIMESTAMP AS timestamp,
    data.service_type::VARCHAR(20) AS service_type,
    data.usage_amount::DECIMAL(10,2) AS usage_amount,
    data.unit::VARCHAR(20) AS unit,
    data.cost_idr::INTEGER AS cost_idr,
    data.location::VARCHAR(50) AS location,
    data.cell_id::VARCHAR(20) AS cell_id,
    data.session_id::VARCHAR(20) AS session_id
FROM 
    demo_kafka_mv;

-- Query the parsed data
SELECT * FROM telco_data_parsed_onprem;

(Optional) Perform real-time analytics while the Python script continues generating events:

-- Real-time usage and cost analysis by subscriber and service type
SELECT
    subscriber_id,
    service_type,
    SUM(usage_amount) as total_usage,
    SUM(cost_idr) as total_cost
FROM telco_data_parsed_onprem
GROUP BY subscriber_id, service_type
ORDER BY subscriber_id, service_type;

Conclusion

In this post, we have demonstrate how to:

Set up a self-managed Kafka cluster on EC2 using Docker with SSL/TLS configuration
Configure DNS resolution using Route53 Private Hosted Zones for secure connectivity
Establish real-time streaming ingestion from Kafka to Amazon Redshift
Create materialized views that automatically refresh with incoming Kafka data
Perform real-time analytics on streaming telco usage data

This architecture enables you to build scalable, real-time data pipelines that can ingest high-volume streaming data directly to Redshift for immediate analysis and reporting.

For more information about Redshift Streaming Ingestion visit the AWS Documentation.

Clean up

To avoid unnecessary charges, clean up all resources created during this demo. Ensure that only delete resources that were created specifically for this demonstration to avoid disrupting other services or applications:

Remove the streaming ingestion components:

-- Drop the materialized view
DROP MATERIALIZED VIEW IF EXISTS demo_kafka_mv;

-- Drop the parsed view (if created)
DROP VIEW IF EXISTS telco_data_parsed_onprem;

-- Drop the external schema
DROP SCHEMA IF EXISTS demo_kafka_onprem;

Terminate EC2 instances
Delete Redshift Cluster/Namespace
Clean up network resources (e.g. Route53 Private Hosted Zone, Security Group)
Remove SSL Certificate (if created specifically for this demo)

Topics: Analytics
Tags: Amazon Redshift Amazon Redshift Serverless
Language: English

EXPERT

Tedy_T

published 9 months ago370 views

No comments

Relevant content

How to Set Up Cross-Account Amazon MSK to Redshift Streaming Over Private Network
EXPERT
Anubhav Gupta
published a year ago
How do I create an Amazon MSK event source mapping to invoke a Lambda function using a NAT Gateway internet connection?
EXPERT
Hitesh
published 2 years ago
Establishing Multi-tenant, Custom Domain-based Secure Connectivity to Amazon MSK through a Kafka Proxy
EXPERT
Ankit Mishra
published 2 years ago
How to measure the lag between AWS Redshift MV and MSK topics in an architecture involving AWS Redshift streaming ingestion from MSK?
MKB
asked 3 years ago
pushing data from MSK to Redshift using EventBridge Pipe
Sukh
asked a year ago
Best Practise and cost effective way to stream data from Kafka topic to Redshift
Accepted Answer
Britto
asked 2 years ago
How do I use the Kafka-Kinesis-Connector to connect to my Amazon MSK cluster?
AWS OFFICIALUpdated 3 years ago
How do I transfer data between Amazon MSK clusters in different accounts with MirrorMaker 2 that runs on MSK Connect?
AWS OFFICIALUpdated 10 months ago
How do I configure the retention period for my MSK cluster?
AWS OFFICIALUpdated a year ago
How do I troubleshoot connectivity issues with Amazon MSK clusters and MSK Connect configurations?
AWS OFFICIALUpdated 3 months ago