Description

Getting Started

Dependencies

Building

Using Maven

# Clone the repositorygit clone<repository-url>cd kafka-client-proxy-to-solace# Build the project and its dependenciesmvn clean package# The built JAR will be available at:target/kafka-wireline-proxy-*.jar

Using Gradle

# Clone the repositorygit clone<repository-url>cd kafka-client-proxy-to-solace# Build the project and its dependencies./gradlew clean build# The built JAR will be available at:build/libs/kafka-wireline-proxy-*.jar# Run directly with Gradle (development mode)./gradlew run-proxy --args="getting-started/sample-configs/proxy-example.properties"

Running as JAR

Using Maven-built JAR

# Run the proxy with a properties filejava -jar target/kafka-wireline-proxy-*.jar /path/to/proxy.properties# Example with JVM tuning optionsjava -Xms512m -Xmx2g -XX:+UseG1GC \     -jar target/kafka-wireline-proxy-*.jar \     /path/to/proxy.properties# With custom logging configurationjava -Dlog4j.configurationFile=log4j2.xml \     -jar target/kafka-wireline-proxy-*.jar \     /path/to/proxy.properties

Using Gradle-built JAR

# Run the proxy with a properties filejava -jar build/libs/kafka-wireline-proxy-*.jar /path/to/proxy.properties# Run directly with Gradle (development mode)./gradlew run-proxy --args="getting-started/sample-configs/proxy-example.properties"# Example with JVM tuning optionsjava -Xms512m -Xmx2g -XX:+UseG1GC \     -jar build/libs/kafka-wireline-proxy-*.jar \     /path/to/proxy.properties# With custom logging configurationjava -Dlog4j.configurationFile=log4j2.xml \     -jar build/libs/kafka-wireline-proxy-*.jar \     /path/to/proxy.properties

Building Demo Clients

# Build all projects (proxy and demo clients)mvn clean package# This creates:# - Main proxy JARls target/kafka-wireline-proxy-*.jar# - Demo producer JARls demo-producer/target/kafka-demo-producer-*.jar# - Demo consumer JARls demo-consumer/target/kafka-demo-consumer-*.jar

Running Demo Clients with Gradle

# Build all projects including demo clients./gradlew clean build# Run demo producer (from separate JAR)java -jar demo-producer/target/kafka-demo-producer-*.jar \     --config getting-started/sample-configs/demo-producer.properties \     --topic PRODUCER_TOPIC:test-topic \     --num-records 10# Run demo consumer (from separate JAR)java -jar demo-consumer/target/kafka-demo-consumer-*.jar \     -c getting-started/sample-configs/demo-consumer.properties \     -g test-group \     -t test-topic

Docker Container

Using Pre-built Image

# Pull the latest imagedocker pull ghcr.io/solacelabs/kafka-wireline-proxy:latest# Run container with pre-built imagedocker run -d \  --name kafka-proxy \  -p 9092:9092 \  -p 9094:9094 \  -p 8080:8080 \  -v /path/to/proxy.properties:/app/proxy.properties \  -v /path/to/certs:/app/certs \  ghcr.io/solacelabs/kafka-wireline-proxy:latest

Building Custom Image

# Build Docker image locallydocker build -t kafka-proxy:latest.# Run container with custom imagedocker run -d \  -p 9092:9092 \  -p 9094:9094 \  -v /path/to/proxy.properties:/app/proxy.properties \  -v /path/to/certs:/app/certs \  kafka-proxy:latest

Kubernetes Deployment

AWS EKS Deployment

# Quick deployment overviewcd k8s/aws-eks-deploy./create-aws-security-groups.shkubectl apply -f instance-lb.yamlkubectl apply -f bootstrap-lb.yamlkubectl apply -f proxy-config-map.yamlkubectl apply -f proxy-sts.yaml

Configuration

Kafka Client Listener Configuration

Basic Listener Settings

SSL/TLS Configuration for Kafka Clients

ssl.enabled.protocols=TLSv1.2,TLSv1.3

mTLS (Mutual TLS) Configuration

Solace Event Broker Connection Settings

Basic Solace Connection

Solace SSL/TLS Configuration

Solace mTLS Configuration

Proxy Operational Configuration

Core Proxy Settings

Consumer Scaling Configuration

Kafka Consumer Defaults

Health Check Configuration

Advanced Configuration

Partition Configuration Details

Environment Variable Resolution

# Basic environment variableadvertised.listeners=${env:KAFKA_ADVERTISED_LISTENERS}# With default valuessl.keystore.password=${env:KAFKA_KEYSTORE_PASSWORD:defaultpass}# Kubernetes-specific tokens (resolved automatically)advertised.listeners=PLAINTEXT://${K8S_INTERNAL_HOSTNAME}:9092,SASL_SSL://${K8S_EXTERNAL_HOSTNAME}:9094

Example Configuration Files

Basic Configuration

# Kafka listenerlisteners=PLAINTEXT://0.0.0.0:9092,SASL_SSL://0.0.0.0:9094# SSL configurationssl.keystore.location=/app/keystore.pkcs12ssl.keystore.password=${env:KAFKA_KEYSTORE_PASSWORD}ssl.keystore.type=PKCS12ssl.enabled.protocols=TLSv1.2# Solace connectionsolace.host=tcps://broker.solace.cloud:55443solace.vpn_name=production-vpn# Proxy settingsproxy.separators=._proxy.partitions.per.topic=50proxy.queuename.qualifier=KAFKA-PROXYmessage.max.bytes=5242880# Health checksproxy.healthcheckserver.create=trueproxy.healthcheckserver.port=8080

Production Kubernetes Configuration

# Dynamic listeners for Kuberneteslisteners=PLAINTEXT://0.0.0.0:9092,SASL_SSL://0.0.0.0:9094advertised.listeners=${env:KAFKA_ADVERTISED_LISTENERS}# SSL with mounted certificatesssl.keystore.location=/app/keystoressl.keystore.password=${env:KAFKA_KEYSTORE_PASSWORD}ssl.keystore.type=PKCS12ssl.enabled.protocols=TLSv1.2ssl.client.auth=none# Solace production brokersolace.host=tcps://production.solace.cloud:55443solace.vpn_name=prod-vpnsolace.ssl.validate_certificate=true# Production scalingproxy.request.handler.threads=64proxy.partitions.per.topic=100proxy.max.uncommitted.messages=2000message.max.bytes=10485760# Monitoringproxy.healthcheckserver.create=trueproxy.healthcheckserver.port=8080

Testing

Authentication & Security

Kafka Client Authentication

Solace Authentication

Limitations

Monitoring & Observability

Health Endpoints

# Health check endpoint - returns 200 OK when proxy is healthycurl http://proxy-host:8080/health# Response: HTTP 200 OK with "OK" body when healthy# Response: HTTP 503 Service Unavailable when unhealthy

livenessProbe:httpGet:path:/healthport:8080initialDelaySeconds:30periodSeconds:10timeoutSeconds:5failureThreshold:3readinessProbe:httpGet:path:/healthport:8080initialDelaySeconds:10periodSeconds:5timeoutSeconds:3failureThreshold:2

Logging

<?xml version="1.0" encoding="UTF-8"?><Configurationstatus="WARN">    <Appenders>        <Consolename="Console"target="SYSTEM_OUT">            <PatternLayoutpattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>        </Console>        <Filename="File"fileName="logs/proxy.log">            <PatternLayoutpattern="%d{yyyy-MM-dd HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>        </File>    </Appenders>        <Loggers>        <Loggername="com.solace.kafka.kafkaproxy"level="INFO"/>        <Loggername="com.solace.kafka.kafkaproxy.ProxyReactor"level="DEBUG"/>        <Loggername="com.solacesystems.jcsmp"level="INFO"/>        <Loggername="org.apache.kafka"level="WARN"/>                <Rootlevel="INFO">            <AppenderRefref="Console"/>            <AppenderRefref="File"/>        </Root>    </Loggers></Configuration>

Metrics

Troubleshooting

Common Issues

Debug Configuration

# Enable debug logging (set in log4j2.xml)# Or via system properties:# -Dlog4j2.logger.com.solace.kafka.kafkaproxy.level=DEBUG# Increase health check verbosity# -Dlog4j2.logger.com.solace.kafka.kafkaproxy.HealthCheckServer.level=DEBUG

License

Contributing

Support

Usage

Kafka Client Connection

Producer Connection Example

Propertiesprops =newProperties();props.put("bootstrap.servers","proxy-host:9092");// or proxy-host:9094 for SSLprops.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");// For SSL connectionsprops.put("security.protocol","SASL_SSL");props.put("sasl.mechanism","PLAIN");props.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required " +"username=\"solace-username\" password=\"solace-password\";");KafkaProducer<String,String>producer =newKafkaProducer<>(props);producer.send(newProducerRecord<>("my-topic","key","Hello Solace!"));

Consumer Connection Example

Propertiesprops =newProperties();props.put("bootstrap.servers","proxy-host:9092");props.put("group.id","my-consumer-group");props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");props.put("auto.offset.reset","earliest");// For SSL connections (same as producer)props.put("security.protocol","SASL_SSL");props.put("sasl.mechanism","PLAIN");props.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required " +"username=\"solace-username\" password=\"solace-password\";");KafkaConsumer<String,String>consumer =newKafkaConsumer<>(props);consumer.subscribe(Collections.singletonList("my-topic"));while (true) {ConsumerRecords<String,String>records =consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String,String>record :records) {System.out.println("Received: " +record.value());    }}

Publishing to Solace Topics

Topic Name Conversion

# Configurationproxy.separators=# Kafka topic → Solace topicorders → ordersuser.events → user.eventsmy_topic_name → my_topic_name

# Configurationproxy.separators=._# Kafka topic → Solace topicorders → ordersuser.events → user/eventsmy_topic_name → my/topic/nameinventory.updates.retail → inventory/updates/retail

Publishing Examples

// Kafka producer codeproducer.send(newProducerRecord<>("inventory.updates.retail","product123",orderData));// With proxy.separators=._ this becomes:// - Solace topic: inventory/updates/retail// - Message published to hierarchical topic structure// - Solace consumers can subscribe to://   - inventory/updates/retail (exact match)//   - inventory/updates/> (wildcard - all retail updates)//   - inventory/> (wildcard - all inventory topics)

Message Headers and Properties

// Kafka producer with headersProducerRecord<String,String>record =newProducerRecord<>("orders","order123",orderJson);record.headers().add("source","web-api".getBytes());record.headers().add("priority","high".getBytes());producer.send(record);// Solace message receives these as user properties:// - source: "web-api"// - priority: "high"

Consuming from Solace Queues

Queue Naming Strategy

[qualifier]/[topic]/[consumer-group]

Queue Name Examples

# Configurationproxy.queuename.qualifier=KAFKA-PROXYproxy.separators=._# Consumer connection:# - Topic: user.profile.updates# - Consumer Group: profile-service## Resulting Solace queue: KAFKA-PROXY/user/profile/updates/profile-service

# Configurationproxy.queuename.qualifier=proxy.separators=.# Consumer connection:# - Topic: inventory.alerts# - Consumer Group: warehouse-app## Resulting Solace queue: inventory/alerts/warehouse-app

# Configurationproxy.queuename.is.topicname=trueproxy.separators=._# Consumer connection:# - Topic: system.notifications# - Consumer Group: email-service (ignored)## Resulting Solace queue: system/notifications# Note: All consumer groups for this topic share the same queue

Consumer Group Behavior

// Consumer 1props.put("group.id","order-processors");consumer1.subscribe(Collections.singletonList("orders"));// Consumer 2props.put("group.id","order-processors");consumer2.subscribe(Collections.singletonList("orders"));// Both consumers share the same Solace queue: KAFKA-PROXY/orders/order-processors// Messages are load-balanced between them (competing consumers)

// Group Aprops.put("group.id","audit-service");consumerA.subscribe(Collections.singletonList("orders"));// Group Bprops.put("group.id","analytics-service");consumerB.subscribe(Collections.singletonList("orders"));// Creates separate Solace queues:// - KAFKA-PROXY/orders/audit-service// - KAFKA-PROXY/orders/analytics-service// Both groups receive all messages (broadcast pattern)

Message Flow Patterns

Publish-Subscribe Pattern

// Publisherproducer.send(newProducerRecord<>("notifications.email",emailData));// Multiple subscribers in different groups// Group 1: Email delivery serviceconsumer1.subscribe(Collections.singletonList("notifications.email"));// Group 2: Email analytics serviceconsumer2.subscribe(Collections.singletonList("notifications.email"));// Result: Both groups receive all email notifications

Load Balancing Pattern

// Multiple consumers in same group for load balancingprops.put("group.id","order-processing-workers");// Worker 1consumer1.subscribe(Collections.singletonList("orders"));// Worker 2consumer2.subscribe(Collections.singletonList("orders"));// Worker 3consumer3.subscribe(Collections.singletonList("orders"));// Result: Orders are distributed across the 3 workers

Hierarchical Topic Publishing

# Configuration enables hierarchical topicsproxy.separators=.

// Kafka publishersproducer.send(newProducerRecord<>("events.user.login",loginEvent));producer.send(newProducerRecord<>("events.user.logout",logoutEvent));producer.send(newProducerRecord<>("events.system.startup",startupEvent));// Solace topics created:// - events/user/login// - events/user/logout// - events/system/startup// Solace consumers can subscribe with wildcards:// - events/user/>     (all user events)// - events/>          (all events)// - events/*/login    (login events from any category)

Authentication Flow

// Kafka client configurationprops.put("sasl.jaas.config","org.apache.kafka.common.security.plain.PlainLoginModule required " +"username=\"my-solace-user\" password=\"my-solace-password\";");// These credentials are passed through to Solace broker// User permissions on Solace determine access to topics/queues

Consumer Scaling Considerations

# Allow up to 20 consumers per topic across all consumer groupsproxy.partitions.per.topic=20

Error Handling

Connection Failures

Message Delivery

Security Features

Request Size Protection

# Maximum request size limit (prevents OutOfMemoryError)proxy.max.request.size.bytes=104857600# 100MB default# Automatic rejection of oversized requests# Logs security events for monitoring suspicious activity

SSL/TLS Connection Validation

Security Event Logging

# Examples of security event logs:[SECURITY][Channel 1] OVERSIZED_REQUEST from /192.168.1.100:45123 - size=2147483647, limit=104857600[SECURITY][Channel 2] PLAINTEXT_ON_SSL_PORT from /10.0.1.50:33445 - detected non-TLS traffic on port 9094

Production JVM Security Configuration

# Recommended production JVM settings for security and stabilityjava -Xms512m -Xmx2g -XX:+UseG1GC \     -XX:MaxGCPauseMillis=200 \     -XX:+DisableExplicitGC \     -Djava.security.egd=file:/dev/./urandom \     -jar kafka-wireline-proxy-*.jar proxy.properties