Debezium + Amazon MSK でアウトボックスパターンによるサービス間連携を構築してみた

@Service
class UserService(
    private val userRepository: UserRepository,
    private val outboxEventRepository: OutboxEventRepository,
    private val objectMapper: ObjectMapper,
) {
    fun listUsers(): List<User> {
        return userRepository.findAll()
    }

    @Transactional
    fun register(name: String): User {
        val user = userRepository.save(User(name = name))
        saveOutboxEvent("USER_REGISTERED", user)
        return user
    }

    @Transactional
    fun update(id: Long, name: String): User {
        val user = userRepository.findById(id)
            .orElseThrow { NoSuchElementException("User not found: id=$id") }
        user.name = name
        user.updatedAt = Instant.now()
        saveOutboxEvent("USER_UPDATED", user)
        return user
    }

    @Transactional
    fun delete(id: Long) {
        val user = userRepository.findById(id)
            .orElseThrow { NoSuchElementException("User not found: id=$id") }
        userRepository.delete(user)
        saveOutboxEvent("USER_DELETED", user)
    }

    // User の登録・更新・削除のたびに、イベントを outbox_events テーブルに書き込む
    private fun saveOutboxEvent(type: String, user: User) {
        val data = objectMapper.writeValueAsBytes(
            mapOf(
                "id" to user.id,
                "name" to user.name,
            )
        )
        // outbox_events の主キーと CloudEvents の id に同じ UUID を使う
        val eventId = UUID.randomUUID().toString()
        // eventのpayloadには CloudEvents SDK for Java を利用
        // https://cloudevents.github.io/sdk-java/spring.html
        val cloudEvent = CloudEventBuilder.v1()
            .withId(eventId)
            .withSource(URI.create("/users"))
            .withType(type)
            .withSubject(user.id.toString())
            .withTime(OffsetDateTime.now(ZoneOffset.UTC))
            .withDataContentType("application/json")
            .withData(data)
            .build()
        outboxEventRepository.save(
            OutboxEvent(
                id = eventId,
                aggregateType = "USER",
                aggregateId = user.id.toString(),
                type = type,
                payload = jsonFormat.serialize(cloudEvent).toString(Charsets.UTF_8),
            )
        )
    }

    companion object {
        private val jsonFormat = JsonFormat()
    }
}

@Entity
@Table(name = "outbox_events")
class OutboxEvent(
    @Id
    @Column(columnDefinition = "char(36)")
    var id: String = UUID.randomUUID().toString(),

    @Column(name = "aggregatetype", nullable = false)
    var aggregateType: String,

    @Column(name = "aggregateid", nullable = false)
    var aggregateId: String,

    @Column(name = "type", nullable = false)
    var type: String,

    @Column(nullable = false, columnDefinition = "json")
    var payload: String,

    @Column(nullable = false)
    var createdAt: Instant = Instant.now(),
)

resource "aws_ecs_task_definition" "app" {
  family                   = "${var.name}-app"
  requires_compatibilities = ["FARGATE"]
  network_mode             = "awsvpc"
  cpu                      = 512
  memory                   = 1024
  execution_role_arn       = aws_iam_role.task_execution.arn

  container_definitions = jsonencode([
    {
      name         = "app"
      image        = "${aws_ecr_repository.app.repository_url}:${var.app_image_tag}"
      essential    = true
      portMappings = [{ containerPort = 8080 }]
      environment = [
        { name = "SPRING_DATASOURCE_URL", value = "jdbc:mysql://${aws_rds_cluster.aurora.endpoint}:3306/${var.db_name}" },
        { name = "SPRING_DATASOURCE_USERNAME", value = var.db_username },
        { name = "SPRING_DATASOURCE_PASSWORD", value = var.db_password },
        { name = "SPRING_KAFKA_BOOTSTRAP_SERVERS", value = aws_msk_cluster.main.bootstrap_brokers },
      ]
      logConfiguration = {
        logDriver = "awslogs"
        options = {
          awslogs-group         = aws_cloudwatch_log_group.app.name
          awslogs-region        = var.region
          awslogs-stream-prefix = "app"
        }
      }
    }
  ])
}

resource "aws_db_subnet_group" "aurora" {
  name       = "${var.name}-aurora"
  subnet_ids = aws_subnet.private[*].id
}

resource "aws_rds_cluster_parameter_group" "aurora" {
  name   = "${var.name}-aurora"
  family = "aurora-mysql8.0"

  # Row-based binlog is required by Debezium
  parameter {
    name         = "binlog_format"
    value        = "ROW"
    apply_method = "pending-reboot"
  }

  parameter {
    name         = "binlog_row_image"
    value        = "FULL"
    apply_method = "immediate"
  }
}

resource "aws_rds_cluster" "aurora" {
  cluster_identifier              = "${var.name}-aurora"
  engine                          = "aurora-mysql"
  database_name                   = var.db_name
  master_username                 = var.db_username
  master_password                 = var.db_password
  db_subnet_group_name            = aws_db_subnet_group.aurora.name
  vpc_security_group_ids          = [aws_security_group.aurora.id]
  db_cluster_parameter_group_name = aws_rds_cluster_parameter_group.aurora.name
  skip_final_snapshot             = true

  serverlessv2_scaling_configuration {
    min_capacity = 0.5
    max_capacity = 2
  }
}

resource "aws_rds_cluster_instance" "aurora" {
  identifier         = "${var.name}-aurora-1"
  cluster_identifier = aws_rds_cluster.aurora.id
  engine             = aws_rds_cluster.aurora.engine
  instance_class     = "db.serverless"
}

CALL mysql.rds_set_configuration('binlog retention hours', 24);

Column        |          Type          | Modifiers
--------------+------------------------+-----------
id            | uuid                   | not null
aggregatetype | character varying(255) | not null
aggregateid   | character varying(255) | not null
type          | character varying(255) | not null
payload       | jsonb                  |

create table outbox_events
(
    id            char(36)     not null primary key,
    aggregateid   varchar(255) not null,
    aggregatetype varchar(255) not null,
    type          varchar(255) not null,
    payload       json         not null,
    created_at    datetime(6)  not null
);

resource "aws_ecs_task_definition" "connect" {
  family                   = "${var.name}-connect"
  requires_compatibilities = ["FARGATE"]
  network_mode             = "awsvpc"
  cpu                      = 1024
  memory                   = 2048
  execution_role_arn       = aws_iam_role.task_execution.arn

  container_definitions = jsonencode([
    {
      name         = "connect"
      image        = "quay.io/debezium/connect:3.6.0.Final"
      essential    = true
      portMappings = [{ containerPort = 8083 }]
      environment = [
        { name = "BOOTSTRAP_SERVERS", value = aws_msk_cluster.main.bootstrap_brokers },
        { name = "GROUP_ID", value = "outbox-connect" },
        { name = "CONFIG_STORAGE_TOPIC", value = "connect_configs" },
        { name = "OFFSET_STORAGE_TOPIC", value = "connect_offsets" },
        { name = "STATUS_STORAGE_TOPIC", value = "connect_statuses" },
      ]
      logConfiguration = {
        logDriver = "awslogs"
        options = {
          awslogs-group         = aws_cloudwatch_log_group.connect.name
          awslogs-region        = var.region
          awslogs-stream-prefix = "connect"
        }
      }
    }
  ])
}

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "REPLACED_BY_AURORA_ENDPOINT",
    "database.port": "3306",
    "database.user": "root",
    "database.password": "password",
    "database.server.id": "184054",
    "topic.prefix": "outbox-app",
    "database.include.list": "test",
    "table.include.list": "test.outbox_events",
    "schema.history.internal.kafka.bootstrap.servers": "REPLACED_BY_MSK_BROKERS",
    "schema.history.internal.kafka.topic": "schema-changes.outbox",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "transforms": "outbox,cloudEventsContentType",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.table.fields.additional.placement": "type:header:eventType",
    "transforms.outbox.table.expand.json.payload": "true",
    "transforms.cloudEventsContentType.type": "org.apache.kafka.connect.transforms.InsertHeader",
    "transforms.cloudEventsContentType.header": "content-type",
    "transforms.cloudEventsContentType.value.literal": "application/cloudevents+json"
  }
}

  "schema.history.internal.kafka.bootstrap.servers": "REPLACED_BY_MSK_BROKERS",
  "schema.history.internal.kafka.topic": "schema-changes.outbox",

resource "aws_msk_configuration" "main" {
  name           = "${var.name}-msk"
  kafka_versions = ["3.9.x"]

  # Debezium and Kafka Connect create topics on demand
  server_properties = <<-PROPERTIES
    auto.create.topics.enable=true
    default.replication.factor=3
    min.insync.replicas=2
  PROPERTIES
}

resource "aws_msk_cluster" "main" {
  cluster_name           = "${var.name}-msk"
  kafka_version          = "3.9.x"
  number_of_broker_nodes = 3

  broker_node_group_info {
    instance_type   = "kafka.t3.small"
    client_subnets  = aws_subnet.private[*].id
    security_groups = [aws_security_group.msk.id]
    storage_info {
      ebs_storage_info {
        volume_size = 20
      }
    }
  }

  configuration_info {
    arn      = aws_msk_configuration.main.arn
    revision = aws_msk_configuration.main.latest_revision
  }

  # Plaintext inside the VPC to keep app / connect config simple (demo only)
  encryption_info {
    encryption_in_transit {
      client_broker = "PLAINTEXT"
      in_cluster    = true
    }
  }
}

@Service
class KafkaEventService {
...
    @KafkaListener(topics = ["outbox.event.USER"], groupId = "consumer-group-id")
    fun onMessage(record: ConsumerRecord<String, String>, @Header("eventType", required = false) eventType: String?) {
        // eventType ヘッダーを使って、イベント種別をログに出力
        log.info("Received event: {}", eventType)
        val event = try {
            // CloudEvents SDK for Java の JsonFormat でパース
            jsonFormat.deserialize(record.value().toByteArray())
        } catch (e: Exception) {
            log.warn("Skipping non-CloudEvent message at offset {}: {}", record.offset(), e.message)
            return
        }
        broadcast(event)
    }
}

{
  "specversion": "1.0",
  "id": "f1bfa6e0-173c-4a31-b3e0-66439bd90468",
  "source": "/users",
  "type": "USER_REGISTERED",
  "datacontenttype": "application/json",
  "subject": "4",
  "time": "2026-07-30T08:45:35.263905545Z",
  "data": {
    "id": 4,
    "name": "Seiichi Arai"
  }
}