diff --git a/.github/actions/test-app-startup/action.yml b/.github/actions/test-app-startup/action.yml
index 21057e04..e0cb1a44 100644
--- a/.github/actions/test-app-startup/action.yml
+++ b/.github/actions/test-app-startup/action.yml
@@ -22,6 +22,8 @@ runs:
     - name: Test application startup
       shell: bash
       working-directory: ${{ inputs.working-directory }}
+      env:
+        KAFKA_BOOTSTRAP_SERVERS: localhost:9092
       run: |
         for service in ${{ inputs.services }}; do
           echo "Testing Service... [$service]"
diff --git a/config/kafka-topics.yml b/config/kafka-topics.yml
index e0b3f031..a8b0bad5 100644
--- a/config/kafka-topics.yml
+++ b/config/kafka-topics.yml
@@ -1,11 +1,12 @@
 spring:
   kafka:
-    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
-  
+    bootstrap-servers: ${KAFKA_BOOTSTRAP_SERVERS:kafka1:29092,kafka2:29092,kafka3:29092}
+    
     listener:
       ack-mode: MANUAL
 
     producer:
+      compression.type: lz4
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
       acks: all
@@ -24,9 +25,11 @@ spring:
       enable-auto-commit: false
       auto-offset-reset: earliest
       max-poll-records: 50
-      max.poll.interval.ms: 300000
-      session.timeout.ms: 15000
-      heartbeat.interval.ms: 3000
+      max-poll-interval-ms: 300000
+      session-timeout-ms: 15000
+      heartbeat-interval-ms: 3000
+      properties:
+        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
       group:
         order: order-group
         payment: payment-group
diff --git a/connectors/order-outbox.json b/connectors/order-outbox.json
index 1d40ef9c..166e39cf 100644
--- a/connectors/order-outbox.json
+++ b/connectors/order-outbox.json
@@ -23,10 +23,6 @@
     "transforms.outbox.table.field.event.payload": "payload",
     "transforms.outbox.route.by.field": "event_type",
     "transforms.outbox.route.topic.replacement": "${routedByValue}",
-    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
-    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
-    "key.converter.schemas.enable": "false",
-    "value.converter.schemas.enable": "false",
     "transforms.outbox.table.expand.json.payload": "true",
     "producer.acks": "all",
     "producer.enable.idempotence": "true",
diff --git a/connectors/payment-outbox.json b/connectors/payment-outbox.json
index 0675d4a9..1ccc0ff8 100644
--- a/connectors/payment-outbox.json
+++ b/connectors/payment-outbox.json
@@ -23,10 +23,6 @@
     "transforms.outbox.table.field.event.payload": "payload",
     "transforms.outbox.route.by.field": "event_type",
     "transforms.outbox.route.topic.replacement": "${routedByValue}",
-    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
-    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
-    "key.converter.schemas.enable": "false",
-    "value.converter.schemas.enable": "false",
     "transforms.outbox.table.expand.json.payload": "true",
     "producer.acks": "all",
     "producer.enable.idempotence": "true",
diff --git a/connectors/register-connectors.sh b/connectors/register-connectors.sh
old mode 100644
new mode 100755
index 7c4a201f..1d6e7a7c
--- a/connectors/register-connectors.sh
+++ b/connectors/register-connectors.sh
@@ -1,27 +1,31 @@
 #!/bin/sh
 
-TARGET_URL=${CONNECT_URL:-"http://connect:8083"}
+TARGET_URL=${CONNECT_URL:-"http://connect1:8083"}
 
-echo "Waiting for Kafka Connect..."
-while [ $(curl -s -o /dev/null -w "%{http_code}" $TARGET_URL) -ne 200 ]; do
+echo "Waiting for Kafka Connect at $TARGET_URL..."
+
+until [ $(curl -s -o /dev/null -w "%{http_code}" "$TARGET_URL") -eq 200 ]; do
+  echo "Connect is not ready yet... (waiting 3s)"
   sleep 3
 done
 
-echo "Registering connectors from /configs..."
+echo " Kafka Connect is UP. Starting registration..."
+
 for file in /configs/*.json; do
-  filename=$(basename "$file")
-  echo "Processing $filename..."
-
-  sed -e "s|\${env:DB_HOST}|$DB_HOST|g" \
-      -e "s|\${env:SPRING_DATASOURCE_USERNAME}|$SPRING_DATASOURCE_USERNAME|g" \
-      -e "s|\${env:SPRING_DATASOURCE_PASSWORD}|$SPRING_DATASOURCE_PASSWORD|g" \
-      -e "s|\${env:DB_NAME}|$DB_NAME|g" \
-      "$file" > "/tmp/$filename"
-
-  response=$(curl -s -X POST -H "Content-Type: application/json" \
-            -d @"/tmp/$filename" \
-            $TARGET_URL/connectors)
-  echo "Response for $filename: $response"
+  connector_name=$(jq -r '.name' "$file")
+  config_body=$(jq -c '.config' "$file")
+
+  echo "Processing Connector: $connector_name"
+  response=$(curl -s -o /dev/null -w "%{http_code}" -X PUT \
+       -H "Content-Type: application/json" \
+       -d "$config_body" \
+       "$TARGET_URL/connectors/$connector_name/config")
+
+  if [ "$response" -eq 200 ] || [ "$response" -eq 201 ]; then
+    echo "Successfully registered/updated: $connector_name"
+  else
+    echo "Failed to register $connector_name (HTTP Status: $response)"
+  fi
 done
 
-echo "ALL connectors Created"
\ No newline at end of file
+echo "ALL connectors have been processed!"
\ No newline at end of file
diff --git a/docker-compose.yaml b/docker-compose.yaml
index 1891a746..f13b650d 100644
--- a/docker-compose.yaml
+++ b/docker-compose.yaml
@@ -31,44 +31,102 @@ services:
       - spot-network
 
   # --- Infrastructure: Event Broker (Kafka) ---
-  kafka:
+  kafka1:
     image: apache/kafka:4.0.0
-    container_name: kafka-local
+    container_name: kafka1
     restart: always
+    environment:
+      KAFKA_NODE_ID: 1
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka1:19093,2@kafka2:19093,3@kafka3:19093'
+      KAFKA_LISTENERS: 'INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:19093'
+      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://kafka1:29092,EXTERNAL://localhost:9092'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT'
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      CLUSTER_ID: 'J9Xz7kQPRYyK8VkqH3mW5A'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
+      KAFKA_MIN_INSYNC_REPLICAS: 2
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      KAFKA_NUM_PARTITIONS: 3
+      KAFKA_HEAP_OPTS: '-Xmx256M -Xms256M'
+      KAFKA_LOG4J_ROOT_LOGLEVEL: 'WARN'
+      KAFKA_LOG4J_LOGGERS: 'kafka=WARN,state.change.logger=WARN'
     ports:
       - "9092:9092"
-      - "19093:19093"
+    volumes:
+      - kafka-data-1:/var/lib/kafka/data
+    networks:
+      - spot-network
+      
+  kafka2:
+    image: apache/kafka:4.0.0
+    container_name: kafka2
+    restart: always
     environment:
-      KAFKA_NODE_ID: 1
+      KAFKA_NODE_ID: 2
+      KAFKA_PROCESS_ROLES: 'broker,controller'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka1:19093,2@kafka2:19093,3@kafka3:19093'
+      KAFKA_LISTENERS: 'INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:19093'
+      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://kafka2:29092,EXTERNAL://localhost:9093'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT'
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
+      CLUSTER_ID: 'J9Xz7kQPRYyK8VkqH3mW5A'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
+      KAFKA_MIN_INSYNC_REPLICAS: 2
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      KAFKA_NUM_PARTITIONS: 3
+      KAFKA_HEAP_OPTS: '-Xmx256M -Xms256M'
       KAFKA_LOG4J_ROOT_LOGLEVEL: 'WARN'
       KAFKA_LOG4J_LOGGERS: 'kafka=WARN,state.change.logger=WARN'
+    ports:
+      - "9093:9093"
+    volumes:
+      - kafka-data-2:/var/lib/kafka/data
+    networks:
+      - spot-network
+
+  kafka3:
+    image: apache/kafka:4.0.0
+    container_name: kafka3
+    restart: always
+    environment:
+      KAFKA_NODE_ID: 3
       KAFKA_PROCESS_ROLES: 'broker,controller'
-      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:19093'
+      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka1:19093,2@kafka2:19093,3@kafka3:19093'
       KAFKA_LISTENERS: 'INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:19093'
-      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://kafka:29092,EXTERNAL://localhost:9092'
+      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://kafka3:29092,EXTERNAL://localhost:9094'
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT'
       KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
       CLUSTER_ID: 'J9Xz7kQPRYyK8VkqH3mW5A'
-      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
-      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
-      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
-      KAFKA_MIN_INSYNC_REPLICAS: 1
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
+      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
+      KAFKA_MIN_INSYNC_REPLICAS: 2
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
       KAFKA_NUM_PARTITIONS: 3
-      KAFKA_HEAP_OPTS: '-Xmx512M -Xms512M'
+      KAFKA_HEAP_OPTS: '-Xmx256M -Xms256M'
+      KAFKA_LOG4J_ROOT_LOGLEVEL: 'WARN'
+      KAFKA_LOG4J_LOGGERS: 'kafka=WARN,state.change.logger=WARN'
+    ports:
+      - "9094:9094"
     volumes:
-      - kafka-data:/var/lib/kafka/data
+      - kafka-data-3:/var/lib/kafka/data
     networks:
       - spot-network
         
-  connect:
+  connect1: &connect-common
+    container_name: connect1
     image: quay.io/debezium/connect:3.4.0.Final
-    container_name: connect
-    depends_on:
-      - kafka
-      - db
     ports:
       - "8888:8083"
     env_file:
@@ -76,47 +134,64 @@ services:
     environment:
       CONNECT_CONFIG_PROVIDERS: 'env'
       CONNECT_CONFIG_PROVIDERS_ENV_CLASS: 'org.apache.kafka.common.config.provider.EnvVarConfigProvider'
-      BOOTSTRAP_SERVERS: kafka:29092
-      GROUP_ID: 1
-      CONFIG_STORAGE_TOPIC: my_connect_configs
-      OFFSET_STORAGE_TOPIC: my_connect_offsets
-      STATUS_STORAGE_TOPIC: my_connect_statuses
+      BOOTSTRAP_SERVERS: kafka1:29092,kafka2:29092,kafka3:29092
+      GROUP_ID: "spot-connect-group"
+      CONFIG_STORAGE_TOPIC: "connect_configs"
+      OFFSET_STORAGE_TOPIC: "connect_offsets"
+      STATUS_STORAGE_TOPIC: "connect_status"
+      CONFIG_STORAGE_REPLICATION_FACTOR: 3
+      OFFSET_STORAGE_REPLICATION_FACTOR: 3
+      STATUS_STORAGE_REPLICATION_FACTOR: 3
       KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
       CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
       CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
       LOGGING_LEVEL: 'WARN'
       CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR"
+    depends_on:
+      - kafka1
+      - kafka2
+      - kafka3
+      - db
     networks:
       - spot-network
+
+  connect2:
+    <<: *connect-common
+    container_name: connect2
+    ports:
+      - "8889:8083"
       
   connect-init:
-    image: curlimages/curl:latest
+    image: alpine:latest
     container_name: connect-init
     env_file:
       - .env
     depends_on:
-      - connect
+      - connect1
+      - connect2
     networks:
       - spot-network
     volumes:
       - ./connectors:/configs
-    entrypoint: ["/bin/sh", "/configs/register-connectors.sh"]
+    entrypoint: ["/bin/sh", "-c", "apk add --no-cache curl jq && chmod +x /configs/register-connectors.sh && /configs/register-connectors.sh"]
 
   kafka-ui:
     image: provectuslabs/kafka-ui:latest
-    container_name: kafka-ui-local
+    container_name: kafka-ui
     ports:
       - "8989:8080"
     environment:
-      KAFKA_CLUSTERS_0_NAME: local-spot-cluster
-      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: 'kafka:29092'
-      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: connect
-      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: 'http://connect:8083'
+      KAFKA_CLUSTERS_0_NAME: spot-cluster
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: 'kafka1:29092,kafka2:29092,kafka3:29092'
+      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: connect-cluster
+      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: 'http://connect1:8083'
       LOGGING_LEVEL_ROOT: WARN
       LOGGING_LEVEL_COM_PROVECTUS: WARN
     depends_on:
-      - kafka
+      - kafka1
+      - kafka2
+      - kafka3
     networks:
       - spot-network
 
@@ -192,14 +267,16 @@ services:
     environment:
       - DB_HOST=db
       - SPRING_DATA_REDIS_HOST=redis
-      - KAFKA_BOOTSTRAP_SERVERS=kafka:29092
+      - KAFKA_BOOTSTRAP_SERVERS=kafka1:29092,kafka2:29092,kafka3:29092
       - SPRING_TEMPORAL_CONNECTION_TARGET=temporal:7233
       - LOGGING_LEVEL_ROOT=WARN
     depends_on:
       - db
       - redis
-      - kafka
       - temporal
+      - kafka1
+      - kafka2
+      - kafka3
     networks:
       - spot-network
     env_file:
@@ -219,12 +296,14 @@ services:
     environment:
       - DB_HOST=db
       - SPRING_DATA_REDIS_HOST=redis
-      - KAFKA_BOOTSTRAP_SERVERS=kafka:29092
+      - KAFKA_BOOTSTRAP_SERVERS=kafka1:29092,kafka2:29092,kafka3:29092
       - LOGGING_LEVEL_ROOT=WARN
     depends_on:
       - db
       - redis
-      - kafka
+      - kafka1
+      - kafka2
+      - kafka3
     networks:
       - spot-network
 
@@ -240,7 +319,6 @@ services:
     environment:
       - DB_HOST=db
       - SPRING_DATA_REDIS_HOST=redis
-      - KAFKA_BOOTSTRAP_SERVERS=kafka:29092
       - LOGGING_LEVEL_ROOT=WARN
     depends_on:
       - db
@@ -262,7 +340,6 @@ services:
     environment:
       - DB_HOST=db
       - SPRING_DATA_REDIS_HOST=redis
-      - KAFKA_BOOTSTRAP_SERVERS=kafka:29092
       - LOGGING_LEVEL_ROOT=WARN
     depends_on:
       - db
@@ -277,5 +354,7 @@ networks:
     driver: bridge
 
 volumes:
-  kafka-data:
+  kafka-data-1:
+  kafka-data-2:
+  kafka-data-3:
   postgres_data:
\ No newline at end of file
diff --git a/infra/k8s/apps/spot-order.yaml b/infra/k8s/apps/spot-order.yaml
index 145ae186..353cabb8 100644
--- a/infra/k8s/apps/spot-order.yaml
+++ b/infra/k8s/apps/spot-order.yaml
@@ -44,13 +44,13 @@ spec:
             httpGet:
               path: /actuator/health
               port: 8082
-            initialDelaySeconds: 30
+            initialDelaySeconds: 150
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /actuator/health
               port: 8082
-            initialDelaySeconds: 60
+            initialDelaySeconds: 180
             periodSeconds: 30
       volumes:
         - name: app-config
diff --git a/infra/k8s/apps/spot-payment.yaml b/infra/k8s/apps/spot-payment.yaml
index 59b19fcd..3ec9b6d2 100644
--- a/infra/k8s/apps/spot-payment.yaml
+++ b/infra/k8s/apps/spot-payment.yaml
@@ -44,13 +44,13 @@ spec:
             httpGet:
               path: /actuator/health
               port: 8084
-            initialDelaySeconds: 30
+            initialDelaySeconds: 150
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /actuator/health
               port: 8084
-            initialDelaySeconds: 60
+            initialDelaySeconds: 180
             periodSeconds: 30
       volumes:
         - name: app-config
diff --git a/infra/k8s/apps/spot-user.yaml b/infra/k8s/apps/spot-user.yaml
index d1215c15..7b1f75a6 100644
--- a/infra/k8s/apps/spot-user.yaml
+++ b/infra/k8s/apps/spot-user.yaml
@@ -44,13 +44,13 @@ spec:
             httpGet:
               path: /actuator/health
               port: 8081
-            initialDelaySeconds: 30
+            initialDelaySeconds: 150
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /actuator/health
               port: 8081
-            initialDelaySeconds: 60
+            initialDelaySeconds: 180
             periodSeconds: 30
       volumes:
         - name: app-config
diff --git a/infra/k8s/base/kafka/allow-kafka-ui-netpol.yaml b/infra/k8s/base/kafka/allow-kafka-ui-netpol.yaml
new file mode 100644
index 00000000..06c25c01
--- /dev/null
+++ b/infra/k8s/base/kafka/allow-kafka-ui-netpol.yaml
@@ -0,0 +1,17 @@
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: allow-kafka-ui-to-connect
+  namespace: spot
+spec:
+  podSelector:
+    matchLabels:
+      strimzi.io/kind: KafkaConnect
+  ingress:
+    - from:
+        - podSelector:
+            matchLabels:
+              app: kafka-ui
+      ports:
+        - protocol: TCP
+          port: 8083
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/connectors.yaml b/infra/k8s/base/kafka/connectors.yaml
new file mode 100644
index 00000000..4db633d7
--- /dev/null
+++ b/infra/k8s/base/kafka/connectors.yaml
@@ -0,0 +1,86 @@
+apiVersion: kafka.strimzi.io/v1
+kind: KafkaConnector
+metadata:
+  name: order-outbox-connector
+  namespace: spot
+  labels:
+    # 이전에 만든 KafkaConnect 리소스 이름(spot-connect)과 반드시 일치해야 함
+    strimzi.io/cluster: spot-connect
+spec:
+  class: io.debezium.connector.postgresql.PostgresConnector
+  tasksMax: 1
+  config:
+    database.hostname: "${env:DB_HOST}"
+    database.port: "5432"
+    database.user: "${env:SPRING_DATASOURCE_USERNAME}"
+    database.password: "${env:SPRING_DATASOURCE_PASSWORD}"
+    database.dbname: "${env:DB_NAME}"
+    topic.prefix: "order_outbox_cdc"
+    plugin.name: "pgoutput"
+    slot.name: "order_outbox_slot"
+    snapshot.mode: "no_data"
+    snapshot.locking.mode: "none"
+    table.include.list: "public.p_order_outbox"
+    tombstones.on.delete: "false"
+    transforms: "outbox"
+    transforms.outbox.type: "io.debezium.transforms.outbox.EventRouter"
+    transforms.outbox.table.field.event.id: "id"
+    transforms.outbox.table.field.event.key: "aggregate_id"
+    transforms.outbox.table.field.event.type: "event_type"
+    transforms.outbox.table.field.event.payload: "payload"
+    transforms.outbox.route.by.field: "event_type"
+    transforms.outbox.route.topic.replacement: "${routedByValue}"
+    transforms.outbox.table.expand.json.payload: "true"
+    producer.acks: "all"
+    producer.enable.idempotence: "true"
+    producer.max.in.flight.requests.per.connection: "5"
+    producer.retries: "100"
+    producer.delivery.timeout.ms: "120000"
+    producer.retry.backoff.ms: "500"
+    producer.compression.type: "lz4"
+    producer.linger.ms: "20"
+    producer.batch.size: "65536"
+
+---
+
+apiVersion: kafka.strimzi.io/v1
+kind: KafkaConnector
+metadata:
+  name: payment-outbox-connector
+  namespace: spot
+  labels:
+    strimzi.io/cluster: spot-connect
+spec:
+  class: io.debezium.connector.postgresql.PostgresConnector
+  tasksMax: 1
+  config:
+    database.hostname: "${env:DB_HOST}"
+    database.port: "5432"
+    database.user: "${env:SPRING_DATASOURCE_USERNAME}"
+    database.password: "${env:SPRING_DATASOURCE_PASSWORD}"
+    database.dbname: "${env:DB_NAME}"
+    topic.prefix: "payment_outbox_cdc"
+    plugin.name: "pgoutput"
+    slot.name: "payment_outbox_slot"
+    snapshot.mode: "no_data"
+    snapshot.locking.mode: "none"
+    table.include.list: "public.p_payment_outbox"
+    tombstones.on.delete: "false"
+    transforms: "outbox"
+    transforms.outbox.type: "io.debezium.transforms.outbox.EventRouter"
+    transforms.outbox.table.field.event.id: "id"
+    transforms.outbox.table.field.event.key: "aggregate_id"
+    transforms.outbox.table.field.event.type: "event_type"
+    transforms.outbox.table.field.event.payload: "payload"
+    transforms.outbox.route.by.field: "event_type"
+    transforms.outbox.route.topic.replacement: "${routedByValue}"
+    transforms.outbox.table.expand.json.payload: "true"
+    producer.acks: "all"
+    producer.enable.idempotence: "true"
+    producer.max.in.flight.requests.per.connection: "5"
+    producer.retries: "100"
+    producer.delivery.timeout.ms: "120000"
+    producer.retry.backoff.ms: "500"
+    producer.compression.type: "lz4"
+    producer.linger.ms: "20"
+    producer.batch.size: "65536"
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/dockerfile b/infra/k8s/base/kafka/dockerfile
new file mode 100644
index 00000000..eecace90
--- /dev/null
+++ b/infra/k8s/base/kafka/dockerfile
@@ -0,0 +1,9 @@
+FROM quay.io/strimzi/kafka:latest-kafka-4.0.0
+
+USER root:root
+RUN mkdir -p /opt/kafka/plugins/debezium
+ADD https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/3.4.0.Final/debezium-connector-postgres-3.4.0.Final-plugin.tar.gz /tmp/
+RUN tar -xzf /tmp/debezium-connector-postgres-3.4.0.Final-plugin.tar.gz -C /opt/kafka/plugins/debezium --strip-components=1 && \
+    rm /tmp/debezium-connector-postgres-3.4.0.Final-plugin.tar.gz
+
+USER 1001
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/kafka-connect-init.yaml b/infra/k8s/base/kafka/kafka-connect-init.yaml
deleted file mode 100644
index d4eb3977..00000000
--- a/infra/k8s/base/kafka/kafka-connect-init.yaml
+++ /dev/null
@@ -1,35 +0,0 @@
-apiVersion: batch/v1
-kind: Job
-metadata:
-  name: kafka-connect-init
-  namespace: spot
-spec:
-  ttlSecondsAfterFinished: 60 # 해당 job 성공 후 60초 뒤에 pod 자동 삭제
-  template:
-    spec:
-      restartPolicy: OnFailure
-      containers:
-        - name: kafka-connect-init
-          image: curlimages/curl:latest
-          env:
-            - name: CONNECT_URL
-              value: "http://kafka-connect-svc:8083"
-          envFrom:
-            - secretRef:
-                name: spot-secrets
-          command: ["/bin/sh"]
-          args: ["/configs/register-connectors.sh"]
-          volumeMounts:
-            - name: config-volume
-              mountPath: /configs
-          resources:
-            requests:
-              memory: "32Mi"
-              cpu: "10m"
-            limits:
-              memory: "64Mi"
-              cpu: "50m"
-      volumes:
-        - name: config-volume
-          configMap:
-            name: kafka-connect-init-config
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/kafka-connect.yaml b/infra/k8s/base/kafka/kafka-connect.yaml
index 87684cc8..d22663d7 100644
--- a/infra/k8s/base/kafka/kafka-connect.yaml
+++ b/infra/k8s/base/kafka/kafka-connect.yaml
@@ -1,61 +1,64 @@
-apiVersion: apps/v1
-kind: Deployment
+apiVersion: kafka.strimzi.io/v1
+kind: KafkaConnect
 metadata:
-  name: kafka-connect
+  name: spot-connect
   namespace: spot
+  annotations:
+    strimzi.io/use-connector-resources: "true"
 spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: kafka-connect
+  version: 4.0.0
+  replicas: 2
+  bootstrapServers: spot-cluster-kafka-bootstrap:9092
+  image: spot-registry.localhost:5111/spot-connect-custom:latest
+  
+  groupId: spot-connect-group
+  configStorageTopic: connect_configs
+  offsetStorageTopic: connect_offsets
+  statusStorageTopic: connect_status
+  
+  config:
+    config.storage.replication.factor: 3
+    offset.storage.replication.factor: 3
+    status.storage.replication.factor: 3
+    
+    key.converter: org.apache.kafka.connect.json.JsonConverter
+    value.converter: org.apache.kafka.connect.json.JsonConverter
+    key.converter.schemas.enable: false
+    value.converter.schemas.enable: false
+    config.providers: env
+    config.providers.env.class: org.apache.kafka.common.config.provider.EnvVarConfigProvider
+    log.level: WARN
+
   template:
-    metadata:
-      labels:
-        app: kafka-connect
-    spec:
-      containers:
-        - name: kafka-connect
-          image: quay.io/debezium/connect:3.4.0.Final
-          imagePullPolicy: IfNotPresent
-          ports:
-            - containerPort: 8083
-          env:
-            - name: CONNECT_CONFIG_PROVIDERS
-              value: 'env'
-            - name: CONNECT_CONFIG_PROVIDERS_ENV_CLASS
-              value: 'org.apache.kafka.common.config.provider.EnvVarConfigProvider'
-            - name: CONNECT_BOOTSTRAP_SERVERS
-              value: kafka:9092
-            - name: GROUP_ID
-              value: "1"
-            - name: CONFIG_STORAGE_TOPIC
-              value: my_connect_configs
-            - name: OFFSET_STORAGE_TOPIC
-              value: my_connect_offsets
-            - name: STATUS_STORAGE_TOPIC
-              value: my_connect_statuses
-            - name: KEY_CONVERTER
-              value: org.apache.kafka.connect.json.JsonConverter
-            - name: VALUE_CONVERTER
-              value: org.apache.kafka.connect.json.JsonConverter
-            - name: CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE
-              value: "false"
-            - name: CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE
-              value: "false"
-            - name: LOGGING_LEVEL
-              value: 'WARN'
-            - name: CONNECT_LOG4J_LOGGERS
-              value: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR"
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: kafka-connect-svc
-  namespace: spot
-spec:
-  type: ClusterIP
-  selector:
-    app: kafka-connect
-  ports:
-      - port: 8083
-        targetPort: 8083
\ No newline at end of file
+    pod:
+      enableServiceLinks: false
+    connectContainer:
+      env:
+        - name: DB_HOST
+          valueFrom:
+            secretKeyRef:
+              name: spot-secrets
+              key: DB_HOST
+        - name: DB_NAME
+          valueFrom:
+            secretKeyRef:
+              name: spot-secrets
+              key: DB_NAME
+        - name: SPRING_DATASOURCE_USERNAME
+          valueFrom:
+            secretKeyRef:
+              name: spot-secrets
+              key: SPRING_DATASOURCE_USERNAME
+        - name: SPRING_DATASOURCE_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: spot-secrets
+              key: SPRING_DATASOURCE_PASSWORD
+
+  resources:
+    requests:
+      cpu: "500m"
+      memory: "512Mi"
+    limits:
+      cpu: "1000m"
+      memory: "1Gi"
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/kafka-nodepool.yml b/infra/k8s/base/kafka/kafka-nodepool.yml
new file mode 100644
index 00000000..195b7696
--- /dev/null
+++ b/infra/k8s/base/kafka/kafka-nodepool.yml
@@ -0,0 +1,26 @@
+apiVersion: kafka.strimzi.io/v1
+kind: KafkaNodePool
+metadata:
+  name: kafka-nodes
+  namespace: spot
+  labels:
+    strimzi.io/cluster: spot-cluster
+spec:
+  replicas: 3
+  roles:
+    - broker
+    - controller
+  storage:
+    type: jbod
+    volumes:
+      - id: 0
+        type: persistent-claim
+        size: 1Gi
+        deleteClaim: true
+  resources:
+    requests:
+      memory: 512Mi
+      cpu: 250m
+    limits:
+      memory: 1Gi
+      cpu: 500m
\ No newline at end of file
diff --git a/infra/k8s/base/kafka/kafka-ui.yaml b/infra/k8s/base/kafka/kafka-ui.yaml
index f394d3be..5ab3dab2 100644
--- a/infra/k8s/base/kafka/kafka-ui.yaml
+++ b/infra/k8s/base/kafka/kafka-ui.yaml
@@ -18,15 +18,23 @@ spec:
           image: provectuslabs/kafka-ui:latest
           ports:
             - containerPort: 8080
+          resources:
+            requests:
+              memory: "384Mi"
+              cpu: "100m"
           env:
             - name: KAFKA_CLUSTERS_0_NAME
-              value: local-spot-cluster
+              value: spot-cluster
             - name: KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS
-              value: 'kafka:9092'
+              value: 'spot-cluster-kafka-bootstrap:9092'
+            
             - name: KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME
-              value: connect
+              value: "spot-connect"
             - name: KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS
-              value: 'http://kafka-connect-svc:8083'
+              value: "http://spot-connect-connect-api.spot.svc:8083"
+            - name: KAFKA_CLUSTERS_0_KAFKACONNECT_0_TYPE
+              value: "kafka-connect"
+
             - name: LOGGING_LEVEL_ROOT
               value: WARN
             - name: LOGGING_LEVEL_COM_PROVECTUS
diff --git a/infra/k8s/base/kafka/kafka.yaml b/infra/k8s/base/kafka/kafka.yaml
index 800b81ae..6a082c0f 100644
--- a/infra/k8s/base/kafka/kafka.yaml
+++ b/infra/k8s/base/kafka/kafka.yaml
@@ -1,98 +1,30 @@
-apiVersion: v1
-kind: PersistentVolumeClaim
+apiVersion: kafka.strimzi.io/v1
+kind: Kafka
 metadata:
-  name: kafka-pvc
+  name: spot-cluster
   namespace: spot
 spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: 1Gi
----
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: kafka
-  namespace: spot
-spec:
-  replicas: 1
-  selector:
-    matchLabels:
-      app: kafka
-  template:
-    metadata:
-      labels:
-        app: kafka
-    spec:
-      enableServiceLinks: false
-      containers:
-        - name: kafka
-          image: apache/kafka:4.0.0
-          imagePullPolicy: IfNotPresent
-          ports:
-            - containerPort: 9092
-            - containerPort: 29092
-          env:
-            - name: KAFKA_NODE_ID
-              value: "1"
-            - name: KAFKA_PROCESS_ROLES
-              value: "broker,controller"
-            - name: KAFKA_CONTROLLER_QUORUM_VOTERS
-              value: "1@localhost:19093"
-            - name: KAFKA_LISTENERS
-              value: "INTERNAL://0.0.0.0:29092,EXTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:19093"
-            - name: KAFKA_ADVERTISED_LISTENERS
-              value: "INTERNAL://kafka:29092,EXTERNAL://kafka:9092"
-            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
-              value: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT"
-            - name: KAFKA_INTER_BROKER_LISTENER_NAME
-              value: "INTERNAL"
-            - name: KAFKA_CONTROLLER_LISTENER_NAMES
-              value: "CONTROLLER"
-            - name: CLUSTER_ID
-              value: "J9Xz7kQPRYyK8VkqH3mW5A"
-            - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
-              value: "1"
-            - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
-              value: "1"
-            - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
-              value: "1"
-            - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
-              value: "true"
-            - name: KAFKA_NUM_PARTITIONS
-              value: "3"
-            - name: KAFKA_HEAP_OPTS
-              value: "-Xmx512M -Xms512M"
-            - name: KAFKA_LOG4J_ROOT_LOGLEVEL
-              value: "WARN"
-          volumeMounts:
-            - name: kafka-storage
-              mountPath: /var/lib/kafka/data
-          resources:
-            requests:
-              memory: "512Mi"
-              cpu: "250m"
-            limits:
-              memory: "1Gi"
-              cpu: "500m"
-      volumes:
-        - name: kafka-storage
-          persistentVolumeClaim:
-            claimName: kafka-pvc
----
-apiVersion: v1
-kind: Service
-metadata:
-  name: kafka
-  namespace: spot
-spec:
-  selector:
-    app: kafka
-  ports:
-    - name: external
-      port: 9092
-      targetPort: 9092
-    - name: internal
-      port: 29092
-      targetPort: 29092
\ No newline at end of file
+  kafka:
+    version: 4.0.0
+    metadataVersion: "3.7-IV4"
+    listeners:
+      - name: plain
+        port: 9092
+        type: internal
+        tls: false
+      - name: controller
+        port: 9093
+        type: internal
+        tls: false
+    config:
+      process.roles: "broker,controller"
+      offsets.topic.replication.factor: 3
+      transaction.state.log.replication.factor: 3
+      transaction.state.log.min.isr: 2
+      default.replication.factor: 3
+      min.insync.replicas: 2
+      auto.create.topics.enable: "true"
+      num.partitions: 3
+  entityOperator:
+    topicOperator: {}
+    userOperator: {}
diff --git a/infra/k8s/base/temporal/temporal-ui.yaml b/infra/k8s/base/temporal/temporal-ui.yaml
index e90b33dc..fc642a9a 100644
--- a/infra/k8s/base/temporal/temporal-ui.yaml
+++ b/infra/k8s/base/temporal/temporal-ui.yaml
@@ -20,7 +20,7 @@ spec:
             - containerPort: 8080
           env:
             - name: TEMPORAL_ADDRESS
-              value: "temporal-svc:7233"
+              value: "temporal:7233"
             - name: TEMPORAL_CORS_ALLOW_ORIGINS
               value: "*"
 ---
diff --git a/infra/k8s/base/temporal/temporal.yaml b/infra/k8s/base/temporal/temporal.yaml
index 5c5194e9..ab115462 100644
--- a/infra/k8s/base/temporal/temporal.yaml
+++ b/infra/k8s/base/temporal/temporal.yaml
@@ -23,7 +23,7 @@ spec:
             - name: DB
               value: postgres12
             - name: POSTGRES_SEEDS
-              value: postgres
+              value: host.k3d.internal
             - name: DB_PORT
               value: "5432"
             - name: POSTGRES_USER
@@ -56,7 +56,7 @@ spec:
 apiVersion: v1
 kind: Service
 metadata:
-  name: temporal-svc
+  name: temporal
   namespace: spot
 spec:
   type: ClusterIP
diff --git a/infra/k8s/kustomization.yaml b/infra/k8s/kustomization.yaml
index 82758521..51a2819a 100644
--- a/infra/k8s/kustomization.yaml
+++ b/infra/k8s/kustomization.yaml
@@ -7,14 +7,16 @@ resources:
   # Base
   - base/namespace.yaml
   - base/configmap.yaml
-  - base/postgres.yaml
-  - base/redis.yaml
+#  - base/postgres.yaml
+#  - base/redis.yaml
 
   # Kafka
   - base/kafka/kafka.yaml
+  - base/kafka/kafka-nodepool.yml
   - base/kafka/kafka-connect.yaml
-  - base/kafka/kafka-connect-init.yaml
+  - base/kafka/connectors.yaml
   - base/kafka/kafka-ui.yaml
+  - base/kafka/allow-kafka-ui-netpol.yaml
 
   # temporal
   - base/temporal/temporal.yaml
diff --git a/run_k3d.sh b/run_k3d.sh
index be6b3f81..8e4429b1 100755
--- a/run_k3d.sh
+++ b/run_k3d.sh
@@ -66,6 +66,9 @@ cleanup_existing() {
     if [ -f "$SCRIPT_DIR/docker-compose.yaml" ]; then
         docker compose -f "$SCRIPT_DIR/docker-compose.yaml" down --remove-orphans 2>/dev/null || true
     fi
+    
+    log_info "Starting essential infrastructure (DB, Redis) via Docker Compose..."
+    docker compose -f "$SCRIPT_DIR/docker-compose.yaml" up -d db redis
 
     if k3d cluster list | grep -q "$CLUSTER_NAME"; then
         log_info "Deleting existing k3d cluster: $CLUSTER_NAME"
@@ -92,19 +95,33 @@ build_and_push_images() {
     log_info "Building and pushing Docker images to local registry..."
 
     SERVICES=("spot-gateway" "spot-user" "spot-store" "spot-order" "spot-payment")
-
+  
+    log_info "Building Kafka Connect with Debezium..."
+    docker build -t "$REGISTRY_NAME:$REGISTRY_PORT/spot-connect-custom:latest" "$SCRIPT_DIR/infra/k8s/base/kafka/"
+    n=0
+    until [ $n -ge 3 ]; do
+        docker push "$REGISTRY_NAME:$REGISTRY_PORT/spot-connect-custom:latest" && break
+        n=$((n+1))
+        log_warn "Push failed for spot-connect-custom. Retrying ($n/3)..."
+        sleep 2
+    done
+    
     for service in "${SERVICES[@]}"; do
         log_info "Building $service..."
-
         (cd "$SCRIPT_DIR/$service" && ./gradlew bootJar -x test)
-
+        
         docker build -t "$REGISTRY_NAME:$REGISTRY_PORT/$service:latest" "$SCRIPT_DIR/$service"
 
-        docker push "$REGISTRY_NAME:$REGISTRY_PORT/$service:latest"
-
+        n=0
+        until [ $n -ge 3 ]; do
+            docker push "$REGISTRY_NAME:$REGISTRY_PORT/$service:latest" && break
+            n=$((n+1))
+            log_warn "Push failed for $service. Retrying ($n/3)..."
+            sleep 2
+        done
         log_info "$service image pushed successfully!"
     done
-}
+  }
 
 install_argocd() {
     log_info "Installing ArgoCD..."
@@ -146,24 +163,47 @@ install_prometheus() {
     log_info "Prometheus installed successfully!"
 }
 
+install_strimzi() {
+  log_info "Installing Strimzi Kafka Operator via Helm..."
+  
+  kubectl create namespace strimzi --dry-run=client -o yaml | kubectl apply -f -
+  kubectl create namespace spot --dry-run=client -o yaml | kubectl apply -f -
+  
+  helm repo add strimzi https://strimzi.io/charts/ >/dev/null 2>&1 || true
+  helm repo update
+
+  helm upgrade --install strimzi-operator strimzi/strimzi-kafka-operator \
+    -n strimzi \
+    --set watchNamespaces={spot} \
+    --wait
+  
+  log_info "Strimzi Operator installed successfully!"
+}
+
 deploy_all() {
     log_info "Deploying all resources using Kustomize..."
 
     # Ingress Controller 설치
     kubectl apply -f https://raw.githubusercontent.com/kubernetes/ingress-nginx/controller-v1.14.3/deploy/static/provider/cloud/deploy.yaml
-
-    # Ingress Controller 대기
     kubectl wait --for=condition=ready pod -n ingress-nginx --selector=app.kubernetes.io/component=controller --timeout=120s
 
-    # Apply all resources using Kustomize (--load-restrictor: 상위 디렉토리 파일 접근 허용)
+    # Kustomize 배포
     kustomize build "$SCRIPT_DIR/infra/k8s/" --load-restrictor LoadRestrictionsNone | kubectl apply -f -
 
-    log_info "Waiting for infrastructure to be ready..."
-    kubectl wait --for=condition=available deployment/postgres -n spot --timeout=180s
-    kubectl wait --for=condition=available deployment/redis -n spot --timeout=180s
-    kubectl wait --for=condition=available deployment/kafka -n spot --timeout=180s
-    kubectl wait --for=condition=available deployment/kafka-connect -n spot --timeout=180s
+#    log_info "Waiting for infrastructure to be ready..."
+#    kubectl wait --for=condition=available deployment/postgres -n spot --timeout=180s
+#    kubectl wait --for=condition=available deployment/redis -n spot --timeout=180s
+    
+    log_info "Waiting for Kafka Cluster (KRaft)..."
+    kubectl wait --for=condition=Ready kafka/spot-cluster -n spot --timeout=300s
+
+    log_info "Waiting for Kafka Connect..."
+    kubectl wait --for=condition=Ready kafkaconnect/spot-connect -n spot --timeout=300s
+
+    log_info "Waiting for Kafka UI..."
     kubectl wait --for=condition=available deployment/kafka-ui -n spot --timeout=180s
+  
+    log_info "Waiting for Temporal..."
     kubectl wait --for=condition=available deployment/temporal -n spot --timeout=180s
     kubectl wait --for=condition=available deployment/temporal-ui -n spot --timeout=180s
 
@@ -214,8 +254,12 @@ main() {
             cleanup_existing
             exit 0
             ;;
-        --status)
-            show_status
+        --build_only)
+            build_and_push_images
+            exit 0
+            ;;
+        --deploy-only)
+            deploy_all
             exit 0
             ;;
     esac
@@ -226,6 +270,7 @@ main() {
     build_and_push_images
     install_argocd
     install_prometheus
+    install_strimzi
     deploy_all
     restart_grafana_for_provisioning
     show_status