도메인 이벤트 기반 색인 파이프라인 Kafka 도입

.github/workflows/ci.yml

@@ -32,6 +32,38 @@ jobs:
           --health-timeout 5s
           --health-retries 10
 
+      zookeeper:
+        image: confluentinc/cp-zookeeper:7.6.1
+        ports:
+          - 2181:2181
+        env:
+          ZOOKEEPER_CLIENT_PORT: 2181
+          ZOOKEEPER_TICK_TIME: 2000
+        options: >-
+          --health-cmd "echo ruok | nc localhost 2181 | grep imok || exit 1"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
+      kafka:
+        image: confluentinc/cp-kafka:7.6.1
+        ports:
+          - 9092:9092
+        env:
+          KAFKA_BROKER_ID: 1
+          KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+          KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+          KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9092
+          KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+          KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+        options: >-
+          --health-cmd "nc -z localhost 9092"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+        depends_on:
+          - zookeeper
+
     steps:
       - name: Check out repository
         uses: actions/checkout@v4
@@ -57,39 +89,29 @@ jobs:
 
       - name: Validate Gradle wrapper
         uses: gradle/wrapper-validation-action@v2
-        with:
-          min-wrapper-count: 1
-          allow-snapshots: false
 
       - name: Make gradlew executable
         run: chmod +x ./gradlew
 
-      - name: Verify Gradle Wrapper Checksum
-        run: |
-          if [ -f "gradle/wrapper/gradle-wrapper.properties" ]; then
-            echo "Gradle wrapper configuration verified"
-            cat gradle/wrapper/gradle-wrapper.properties
-          fi
-
       - name: Setup Gradle
         uses: gradle/gradle-build-action@v2
         with:
           gradle-build-scan-report: true
           cache-read-only: false
 
-      - name: Wait for Elasticsearch to be ready
+      - name: Wait for services to be ready
         run: |
-          echo "Waiting for Elasticsearch..."
+          echo "Waiting for Elasticsearch, Kafka, and Zookeeper..."
           for i in {1..20}; do
-            if curl -fsS http://localhost:9200 >/dev/null; then
-              echo "Elasticsearch is up!"
+            if curl -fsS http://localhost:9200 >/dev/null && nc -z localhost 9092 && nc -z localhost 2181; then
+              echo "All services are up!"
               exit 0
             fi
-              echo "Retrying in 3s..."
-              sleep 3
-            done
-            echo "Elasticsearch did not become ready in time."
-            exit 1
+            echo "Retrying in 5s..."
+            sleep 5
+          done
+          echo "Services did not become ready in time."
+          exit 1
 
       - name: Run tests
         run: |
@@ -102,6 +124,7 @@ jobs:
             -Dspring.profiles.active=test
         env:
           SPRING_ELASTICSEARCH_URIS: http://localhost:9200
+          SPRING_KAFKA_BOOTSTRAP_SERVERS: localhost:9092
 
       - name: Upload test results
         uses: actions/upload-artifact@v4
@@ -111,36 +134,4 @@ jobs:
           path: |
             build/reports/tests/
             build/test-results/
-          retention-days: 7
-
-  build-check:
-    runs-on: ubuntu-latest
-    if: github.event_name == 'pull_request'
-
-    steps:
-      - name: Check out repository
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 1
-
-      - name: Set up JDK 21
-        uses: actions/setup-java@v4
-        with:
-          java-version: '21'
-          distribution: 'temurin'
-          cache: gradle
-
-      - name: Create application.yml file
-        run: |
-          mkdir -p ./src/main/resources
-          echo "${{ secrets.APPLICATION_YML }}" > ./src/main/resources/application.yml
-
-      - name: Setup Gradle
-        uses: gradle/gradle-build-action@v2
-
-      - name: Compile check
-        run: |
-          ./gradlew compileJava compileTestJava \
-            --parallel \
-            --build-cache \
-            --configuration-cache
+          retention-days: 7

build.gradle.kts

@@ -48,6 +48,7 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-web")
     implementation("org.springframework.boot:spring-boot-starter-validation")
     implementation("org.springframework.security:spring-security-core")
+    implementation("org.springframework.kafka:spring-kafka")
     implementation("io.jsonwebtoken:jjwt-api:0.12.6")
     implementation("io.jsonwebtoken:jjwt-impl:0.12.6")
     implementation("io.jsonwebtoken:jjwt-jackson:0.12.6")
@@ -62,6 +63,7 @@ dependencies {
 
     testCompileOnly("org.projectlombok:lombok")
     testImplementation("org.springframework.boot:spring-boot-starter-test")
+    testImplementation("org.springframework.kafka:spring-kafka-test")
     testRuntimeOnly("org.junit.platform:junit-platform-launcher")
     testImplementation("org.junit-pioneer:junit-pioneer:2.3.0")
     testImplementation("org.mockito:mockito-core:5.18.0")

docker-compose.yml

@@ -1,7 +1,10 @@
+version: "3.8"
+
 services:
   mysql:
     image: mysql:8.0
     container_name: see-mysql
+    restart: unless-stopped
     ports:
       - "3306:3306"
     environment:
@@ -14,30 +17,15 @@ services:
       - ./docker/mysql/init:/docker-entrypoint-initdb.d
     command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
     healthcheck:
-      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
-      timeout: 20s
+      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-proot"]
+      interval: 10s
+      timeout: 5s
       retries: 10
-    restart: unless-stopped
-
-#  mysql-test:
-#    image: mysql:8.0
-#    container_name: see-test-mysql
-#    ports:
-#      - "3307:3306"
-#    environment:
-#      MYSQL_ROOT_PASSWORD: root
-#      MYSQL_DATABASE: see_test
-#      MYSQL_USER: see_user
-#      MYSQL_PASSWORD: see_password
-#    volumes:
-#      - mysql_test_data:/var/lib/mysql
-#      - ./docker/mysql/init:/docker-entrypoint-initdb.d
-#    command: --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
-#    restart: unless-stopped
 
   elasticsearch:
     image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
     container_name: see-elasticsearch
+    restart: unless-stopped
     environment:
       - discovery.type=single-node
       - xpack.security.enabled=false
@@ -51,23 +39,64 @@ services:
       interval: 10s
       timeout: 5s
       retries: 5
-    restart: unless-stopped
 
   kibana:
     image: docker.elastic.co/kibana/kibana:8.12.0
     container_name: see-kibana
+    restart: unless-stopped
     ports:
       - "5601:5601"
     environment:
       ELASTICSEARCH_HOSTS: http://see-elasticsearch:9200
     depends_on:
-      - elasticsearch
+      elasticsearch:
+        condition: service_healthy
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.6.1
+    container_name: see-zookeeper
     restart: unless-stopped
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "2181:2181"
+    healthcheck:
+      test: ["CMD-SHELL", "echo ruok | nc localhost 2181 | grep imok || exit 1"]
+      interval: 10s
+      timeout: 5s
+      retries: 10
+
+  kafka:
+    image: confluentinc/cp-kafka:7.6.1
+    container_name: see-kafka
+    restart: unless-stopped
+    depends_on:
+      zookeeper:
+        condition: service_healthy
+    ports:
+      - "9092:9092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+
+      # 두 가지 리스너를 노출하여 컨테이너 내부/호스트 접근 모두 지원
+      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:29092,PLAINTEXT_HOST://0.0.0.0:9092
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://see-kafka:29092,PLAINTEXT_HOST://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+
+      # 기본 동작 설정
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_LOG_RETENTION_HOURS: 168
+      KAFKA_DELETE_TOPIC_ENABLE: "true"
+
+    healthcheck:
+      test: [ "CMD", "bash", "-c", "kafka-topics --bootstrap-server localhost:9092 --list >/dev/null 2>&1" ]
+      interval: 10s
+      timeout: 5s
+      retries: 10
 
 volumes:
   mysql_data:
-    driver: local
-  mysql_test_data:
   es_data:
-    driver: local
-

src/main/java/dooya/see/adapter/integration/kafka/DirectPostEventPublisher.java

@@ -0,0 +1,22 @@
+package dooya.see.adapter.integration.kafka;
+
+import dooya.see.application.post.required.PostEventPublisher;
+import dooya.see.domain.shared.DomainEvent;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty(value = "see.kafka.enabled", havingValue = "false", matchIfMissing = true)
+public class DirectPostEventPublisher implements PostEventPublisher {
+    private final PostEventProcessor postEventProcessor;
+
+    @Override
+    public void publish(DomainEvent event) {
+        log.debug("Kafka 비활성화 상태 - 도메인 이벤트를 직접 처리합니다: {}", event);
+        postEventProcessor.process(event);
+    }
+}

src/main/java/dooya/see/adapter/integration/kafka/PostEventConsumer.java

@@ -0,0 +1,24 @@
+package dooya.see.adapter.integration.kafka;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+@ConditionalOnProperty(value = "see.kafka.enabled", havingValue = "true")
+public class PostEventConsumer {
+    private final PostEventProcessor postEventProcessor;
+
+    @KafkaListener(
+            topics = "${see.kafka.topics.post-events:post-events}",
+            groupId = "${spring.kafka.consumer.group-id:post-indexer-group}"
+    )
+    public void consume(PostEventMessage message) {
+        log.debug("Kafka 이벤트 수신: {}", message);
+        postEventProcessor.process(message);
+    }
+}

src/main/java/dooya/see/adapter/integration/kafka/PostEventMessage.java

@@ -0,0 +1,17 @@
+package dooya.see.adapter.integration.kafka;
+
+import java.io.Serializable;
+
+public record PostEventMessage(PostEventType type, Long postId) implements Serializable {
+    static PostEventMessage created(Long postId) {
+        return new PostEventMessage(PostEventType.CREATED, postId);
+    }
+
+    static PostEventMessage updated(Long postId) {
+        return new PostEventMessage(PostEventType.UPDATED, postId);
+    }
+
+    static PostEventMessage deleted(Long postId) {
+        return new PostEventMessage(PostEventType.DELETED, postId);
+    }
+}