Merge pull request #36 from usdot-jpo-ode/jikkou-updates

Confluent Cloud Topic Creation
usdot-jpo-ode · Feb 6, 2025 · 8ba9dc2 · 8ba9dc2
2 parents f99374d + 94e4f13
commit 8ba9dc2
Show file tree

Hide file tree

Showing 5 changed files with 37 additions and 10 deletions.
diff --git a/README.md b/README.md
@@ -16,6 +16,7 @@ The JPO ITS utilities repository serves as a central location for deploying open
     - [Quick Run](#quick-run)
   - [3. Kafka](#3-kafka)
     - [Configure Topic Creation](#configure-topic-creation)
+      - [Confluent Cloud Support](#confluent-cloud-support)
     - [Quick Run](#quick-run-1)
   - [4. MongoDB Kafka Connect](#4-mongodb-kafka-connect)
     - [Configuration](#configuration)
@@ -106,6 +107,17 @@ The following enviroment variables can be used to configure Kafka Topic creation
 | `KAFKA_TOPIC_RETENTION_MS` | Retention time for stream topics, milliseconds |
 | `KAFKA_TOPIC_DELETE_RETENTION_MS` | Tombstone retention time for compacted topics, milliseconds |
 
+#### Confluent Cloud Support
+
+The following environment variables are used to configure the Kafka client for Confluent Cloud.
+
+| Environment Variable | Description |
+|---|---|
+| `KAFKA_SECURITY_PROTOCOL` | Security protocol for Kafka |
+| `KAFKA_SASL_MECHANISM` | SASL mechanism for Kafka |
+| `KAFKA_SASL_JAAS_CONFIG` | SASL JAAS configuration for Kafka |
+| `KAFKA_SSL_ENDPOINT_ALGORITHM` | SSL endpoint algorithm for Kafka |
+
 ### Quick Run
 
 1. Create a copy of `sample.env` and rename it to `.env`.

diff --git a/docker-compose-kafka.yml b/docker-compose-kafka.yml
@@ -75,6 +75,11 @@ services:
       KAFKA_TOPIC_CREATE_CONFLICTMONITOR: ${KAFKA_TOPIC_CREATE_CONFLICTMONITOR}
       KAFKA_TOPIC_CREATE_DEDUPLICATOR: ${KAFKA_TOPIC_CREATE_DEDUPLICATOR}
       KAFKA_TOPIC_CREATE_MECDEPOSIT: ${KAFKA_TOPIC_CREATE_MECDEPOSIT}
+
+      KAFKA_SECURITY_PROTOCOL: ${KAFKA_SECURITY_PROTOCOL:-PLAINTEXT}
+      KAFKA_SASL_MECHANISM: ${KAFKA_SASL_MECHANISM}
+      KAFKA_SASL_JAAS_CONFIG: ${KAFKA_SASL_JAAS_CONFIG}
+      KAFKA_SSL_ENDPOINT_ALGORITHM: ${KAFKA_SSL_ENDPOINT_ALGORITHM}
     logging:
       options:
         max-size: "10m"

diff --git a/jikkou/application.conf b/jikkou/application.conf
@@ -1,4 +1,3 @@
-
 jikkou {
 
   extension.providers {
@@ -23,13 +22,17 @@ jikkou {
     client {
       bootstrap.servers = "kafka:9092"
       bootstrap.servers = ${?KAFKA_BOOTSTRAP_SERVERS}
-    #   security.protocol = "SSL"
-    #   ssl.keystore.location = "/tmp/client.keystore.p12"
-    #   ssl.keystore.password = "password"
-    #   ssl.keystore.type = "PKCS12"
-    #   ssl.truststore.location = "/tmp/client.truststore.jks"
-    #   ssl.truststore.password = "password"
-    #   ssl.key.password = "password"
+
+      # Security protocol defaults to PLAINTEXT for local development
+      security.protocol = "PLAINTEXT"
+      security.protocol = ${?KAFKA_SECURITY_PROTOCOL}
+
+      # SASL Configuration - only used when security.protocol is SASL_SSL
+      sasl.mechanism = ${?KAFKA_SASL_MECHANISM}
+      sasl.jaas.config = ${?KAFKA_SASL_JAAS_CONFIG}
+
+      # SSL Configuration for Confluent Cloud - only used with SASL_SSL
+      ssl.endpoint.identification.algorithm = ${?KAFKA_SSL_ENDPOINT_ALGORITHM}
     }
     brokers {
       # If 'True' 

diff --git a/jikkou/kafka-topics-template.jinja b/jikkou/kafka-topics-template.jinja
@@ -29,8 +29,8 @@ metadata:
   labels: 
     app: "{{ app.name }}"
 spec:
-  partitions: 1
-  replicas: 1
+  partitions: {{ topic.partitions | default(values.partitions) }}
+  replicas: {{ system.env.KAFKA_TOPIC_REPLICAS | default(values.replicas) }}
   configs:
     cleanup.policy: compact
     min.insync.replicas: {{ system.env.KAFKA_TOPIC_MIN_INSYNC_REPLICAS | default(values.minInsyncReplicas) }}

diff --git a/sample.env b/sample.env
@@ -49,6 +49,13 @@ KAFKA_TOPIC_CREATE_GEOJSONCONVERTER=true    # Create topics for GeoJSON Converte
 KAFKA_TOPIC_CREATE_CONFLICTMONITOR=true     # Create topics for Conflict Monitor
 KAFKA_TOPIC_CREATE_DEDUPLICATOR=false       # Create topics for Deduplicator
 KAFKA_TOPIC_CREATE_MECDEPOSIT=false         # Create topics for MecDeposit
+
+# Confluent Cloud Support
+KAFKA_SECURITY_PROTOCOL=PLAINTEXT
+KAFKA_SASL_MECHANISM=PLAIN
+KAFKA_SASL_JAAS_CONFIG=
+KAFKA_SSL_ENDPOINT_ALGORITHM=
+
 ### KAFKA variables - END ###
 
 ### MONGODB variables - START ###