add prod kafka topics flow

.gitignore

@@ -20,6 +20,7 @@ terragrunt.hcl.backup
 
 # Generated files
 .gitlab-ci.generated.yml
+.generated/
 
 # IDE
 .idea/

.gitlab-ci.yml

@@ -25,6 +25,7 @@ generate-pipeline:
       - .gitlab-ci.generated.yml
     expire_in: 1 hour
   rules:
+    - if: '$CI_PIPELINE_SOURCE == "pipeline"'
     - if: '$CI_PIPELINE_SOURCE == "merge_request_event"'
     - if: '$CI_COMMIT_BRANCH == "stage"'
     - if: '$CI_COMMIT_BRANCH == "master"'
@@ -41,6 +42,9 @@ trigger-downstream-prod:
       - artifact: .gitlab-ci.generated.yml
         job: generate-pipeline
     strategy: depend
+    forward:
+      pipeline_variables: true
+      yaml_variables: true
   variables:
     TERRAFORM_IMAGE_NAME: cr.yandex/crp3ccidau046kdj8g9q/terraform/terragrunt:v9.10
     YC_SERVICE_ACCOUNT_KEY_FILE: ${YC_PROD_SERVICE_ACCOUNT_KEY_FILE}
@@ -55,8 +59,9 @@ trigger-downstream-prod:
     AWS_SECRET_ACCESS_KEY: ${S3_PROD_SECRET_KEY}
     TARGET_ENV: "prod"
   rules:
+    - if: '$CI_PIPELINE_SOURCE == "pipeline"'
     - if: '$CI_PIPELINE_SOURCE == "merge_request_event"'
     - if: '$CI_COMMIT_BRANCH == "master"'
     - if: '$CI_COMMIT_BRANCH =~ /^feature\/.*/'
 
-##
+##

infrastructure.yaml

@@ -1,5 +1,22 @@
 environments:
   prod:
+    kafka_cluster_refs:
+      prod:
+        cluster_id: c9qks74g084g1419cpts
+        default_partitions: 3
+        default_replication_factor: 3
+        max_replication_factor: 3
+        default_topic_config:
+          cleanup.policy: delete
+          retention.ms: "604800000"
+          min.insync.replicas: "2"
+
+    kafka_policy:
+      allow_create: true
+      allow_delete: false
+      allow_partition_increase: true
+      allow_partition_decrease: false
+
     namespaces:
       - name: pulse
         labels:

live/prod/kafka-topics/terragrunt.hcl

@@ -0,0 +1,73 @@
+include "root" {
+  path = find_in_parent_folders()
+}
+
+include "env" {
+  path           = find_in_parent_folders("env.hcl")
+  expose         = true
+  merge_strategy = "deep"
+}
+
+locals {
+  repo_root    = try(get_repo_root(), "${get_terragrunt_dir()}/../../..")
+  infra_config = yamldecode(file("${local.repo_root}/infrastructure.yaml"))
+  env_name     = basename(dirname(get_terragrunt_dir()))
+  env_config   = local.infra_config.environments[local.env_name]
+
+  manifest_repo = get_env("TOPIC_MANIFEST_REPO", "local")
+  manifest_path = get_env("TOPIC_MANIFEST_PATH", "")
+  manifest_slug = replace(replace(replace(replace("${local.manifest_repo}:${local.manifest_path}", "/", "__"), ":", "__"), ".", "_"), " ", "_")
+
+  topic_manifest_path = get_env(
+    "TOPIC_MANIFEST_LOCAL_PATH",
+    "${local.repo_root}/.generated/kafka-topics/${local.env_name}/manifest.yaml"
+  )
+
+  kafka_cluster_refs = {
+    for ref, cfg in local.env_config.kafka_cluster_refs : ref => merge(cfg, {
+      cluster_id = startswith(tostring(cfg.cluster_id), "env:") ? get_env(trimprefix(tostring(cfg.cluster_id), "env:"), "") : tostring(cfg.cluster_id)
+    })
+  }
+}
+
+terraform {
+  source = "${get_terragrunt_dir()}/../../../modules//kafka-topics-yc"
+}
+
+remote_state {
+  backend = "s3"
+  generate = {
+    path      = "backend.tf"
+    if_exists = "overwrite_terragrunt"
+  }
+  config = {
+    endpoint                    = "https://storage.yandexcloud.net"
+    bucket                      = get_env("S3_STATE_BUCKET", get_env("S3_PROD_STATE_BUCKET", "tfstate-terragrunt-prod"))
+    key                         = "live/${local.env_name}/kafka-topics/${local.manifest_slug}/terraform.tfstate"
+    region                      = "ru-central1"
+    access_key                  = get_env("S3_ACCESS_KEY", get_env("AWS_ACCESS_KEY_ID", ""))
+    secret_key                  = get_env("S3_SECRET_KEY", get_env("AWS_SECRET_ACCESS_KEY", ""))
+    skip_region_validation      = true
+    skip_credentials_validation = true
+    skip_metadata_api_check     = true
+    force_path_style            = true
+    skip_bucket_root_access     = true
+  }
+}
+
+inputs = {
+  environment          = local.env_name
+  topic_manifest_path  = local.topic_manifest_path
+  kafka_cluster_refs   = local.kafka_cluster_refs
+  kafka_policy         = local.env_config.kafka_policy
+  create_users         = true
+  default_user_roles   = ["ACCESS_ROLE_PRODUCER", "ACCESS_ROLE_CONSUMER"]
+  user_password_length = 32
+
+  yc_token                    = get_env("YC_TOKEN", "")
+  yc_service_account_key_file = get_env("YC_SERVICE_ACCOUNT_KEY_FILE", "")
+  yc_cloud_id                 = get_env("YC_CLOUD_ID", "")
+  yc_folder_id                = get_env("YC_PROD_FOLDER_ID", get_env("YC_FOLDER_ID", ""))
+  kubeconfig_path             = get_env("KUBECONFIG", "")
+  kube_context                = get_env("KUBE_CONTEXT", "")
+}

live/terragrunt.hcl

@@ -30,7 +30,7 @@ terraform {
   required_providers {
     yandex = {
       source  = "yandex-cloud/yandex"
-      version = "~> 0.100"
+      version = ">= 0.161.0"
     }
     kubernetes = {
       source  = "hashicorp/kubernetes"

modules/kafka-topics-yc/main.tf

@@ -0,0 +1,128 @@
+locals {
+  manifest = yamldecode(file(var.topic_manifest_path))
+
+  raw_topics = try(local.manifest.kafka.topics, [])
+
+  topics = {
+    for topic in local.raw_topics : topic.name => {
+      name        = topic.name
+      owner       = topic.owner
+      cluster_ref = topic.clusterRef
+      cluster     = var.kafka_cluster_refs[topic.clusterRef]
+
+      partitions = try(
+        tonumber(topic.partitions[var.environment]),
+        try(tonumber(topic.partitions._default), try(tonumber(topic.partitions), var.kafka_cluster_refs[topic.clusterRef].default_partitions))
+      )
+
+      replication_factor = try(
+        tonumber(topic.replicationFactor[var.environment]),
+        try(tonumber(topic.replicationFactor._default), try(tonumber(topic.replicationFactor), var.kafka_cluster_refs[topic.clusterRef].default_replication_factor))
+      )
+
+      config = {
+        for key, value in merge(
+          try(var.kafka_cluster_refs[topic.clusterRef].default_topic_config, {}),
+          try(topic.config, {})
+        ) : key => try(tostring(value[var.environment]), try(tostring(value._default), tostring(value)))
+      }
+
+      deletion_policy = try(topic.deletionPolicy, "orphan")
+    }
+  }
+
+  owner_clusters = {
+    for item in distinct([
+      for topic in values(local.topics) : "${topic.cluster_ref}:${topic.owner}"
+      ]) : item => {
+      cluster_ref = split(":", item)[0]
+      owner       = split(":", item)[1]
+      cluster_id  = var.kafka_cluster_refs[split(":", item)[0]].cluster_id
+    }
+  }
+
+  owner_permissions = {
+    for owner_key, owner in local.owner_clusters : owner_key => flatten([
+      for topic in values(local.topics) : [
+        for role in var.default_user_roles : {
+          topic_name = topic.name
+          role       = role
+        }
+      ]
+      if topic.cluster_ref == owner.cluster_ref && topic.owner == owner.owner
+    ])
+  }
+}
+
+resource "random_password" "kafka_user" {
+  for_each = var.create_users ? local.owner_clusters : {}
+
+  length  = var.user_password_length
+  special = false
+  upper   = true
+  lower   = true
+  numeric = true
+}
+
+resource "yandex_mdb_kafka_topic" "this" {
+  for_each = local.topics
+
+  cluster_id         = each.value.cluster.cluster_id
+  name               = each.value.name
+  partitions         = each.value.partitions
+  replication_factor = each.value.replication_factor
+
+  dynamic "topic_config" {
+    for_each = length(each.value.config) == 0 ? [] : [each.value.config]
+    content {
+      cleanup_policy = (
+        try(topic_config.value["cleanup.policy"], null) == "delete" ? "CLEANUP_POLICY_DELETE" :
+        try(topic_config.value["cleanup.policy"], null) == "compact" ? "CLEANUP_POLICY_COMPACT" :
+        null
+      )
+      compression_type = try(topic_config.value["compression.type"], null)
+      min_insync_replicas = try(
+        tonumber(topic_config.value["min.insync.replicas"]),
+        null
+      )
+      retention_bytes = try(
+        tonumber(topic_config.value["retention.bytes"]),
+        null
+      )
+      retention_ms = try(
+        tonumber(topic_config.value["retention.ms"]),
+        null
+      )
+      segment_bytes = try(
+        tonumber(topic_config.value["segment.bytes"]),
+        null
+      )
+    }
+  }
+
+  lifecycle {
+    prevent_destroy = true
+  }
+}
+
+resource "yandex_mdb_kafka_user" "this" {
+  for_each = var.create_users ? local.owner_clusters : {}
+
+  cluster_id = each.value.cluster_id
+  name       = each.value.owner
+  password   = random_password.kafka_user[each.key].result
+
+  dynamic "permission" {
+    for_each = try(local.owner_permissions[each.key], [])
+    content {
+      topic_name = permission.value.topic_name
+      role       = permission.value.role
+    }
+  }
+
+  lifecycle {
+    prevent_destroy = true
+  }
+
+  depends_on = [yandex_mdb_kafka_topic.this]
+}

modules/kafka-topics-yc/outputs.tf

@@ -0,0 +1,9 @@
+output "topic_names" {
+  description = "Kafka topics managed by this service manifest state."
+  value       = keys(yandex_mdb_kafka_topic.this)
+}
+
+output "user_names" {
+  description = "Kafka users managed by this service manifest state."
+  value       = [for user in yandex_mdb_kafka_user.this : user.name]
+}

modules/kafka-topics-yc/variables.tf

@@ -0,0 +1,49 @@
+variable "environment" {
+  description = "Target environment name, for example stage/preprod/prod."
+  type        = string
+}
+
+variable "topic_manifest_path" {
+  description = "Local path to service-owned Kafka topics manifest."
+  type        = string
+}
+
+variable "kafka_cluster_refs" {
+  description = "Environment clusterRef mapping from infrastructure.yaml."
+  type = map(object({
+    cluster_id                 = string
+    default_partitions         = optional(number, 3)
+    default_replication_factor = optional(number, 1)
+    max_replication_factor     = optional(number, 1)
+    default_topic_config       = optional(map(string), {})
+  }))
+}
+
+variable "kafka_policy" {
+  description = "Platform policy for service-owned Kafka declarations."
+  type = object({
+    allow_create             = optional(bool, true)
+    allow_delete             = optional(bool, false)
+    allow_partition_increase = optional(bool, true)
+    allow_partition_decrease = optional(bool, false)
+  })
+  default = {}
+}
+
+variable "create_users" {
+  description = "Create one Kafka user per topic owner."
+  type        = bool
+  default     = true
+}
+
+variable "default_user_roles" {
+  description = "Kafka roles granted to generated owner users on their topics."
+  type        = list(string)
+  default     = ["ACCESS_ROLE_PRODUCER", "ACCESS_ROLE_CONSUMER"]
+}
+
+variable "user_password_length" {
+  description = "Generated Kafka user password length."
+  type        = number
+  default     = 32
+}

modules/kafka-topics-yc/versions.tf

@@ -0,0 +1,3 @@
+terraform {
+  required_version = ">= 1.5.0"
+}

scripts/fetch-topic-manifest.sh

@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+env_name="${TOPIC_ENV:-${ENVIRONMENT:-stage}}"
+repo_root="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
+output_path="${TOPIC_MANIFEST_LOCAL_PATH:-${repo_root}/.generated/kafka-topics/${env_name}/manifest.yaml}"
+
+mkdir -p "$(dirname "${output_path}")"
+
+if [[ -z "${TOPIC_MANIFEST_PATH:-}" ]]; then
+  echo "TOPIC_MANIFEST_PATH is required for kafka-topics component" >&2
+  exit 1
+fi
+
+if [[ -z "${TOPIC_MANIFEST_REPO:-}" ]]; then
+  if [[ -f "${TOPIC_MANIFEST_PATH}" ]]; then
+    cp "${TOPIC_MANIFEST_PATH}" "${output_path}"
+    echo "Copied local topic manifest to ${output_path}"
+    exit 0
+  fi
+
+  echo "TOPIC_MANIFEST_REPO is required when ${TOPIC_MANIFEST_PATH} is not local" >&2
+  exit 1
+fi
+
+if [[ -z "${TOPIC_MANIFEST_SHA:-}" ]]; then
+  echo "TOPIC_MANIFEST_SHA is required" >&2
+  exit 1
+fi
+
+server_url="${CI_SERVER_URL:-https://gitlab.sarex.io}"
+encoded_path="${TOPIC_MANIFEST_PATH#/}"
+urlencode() {
+  python3 -c 'import sys, urllib.parse; print(urllib.parse.quote(sys.argv[1], safe=""))' "$1"
+}
+
+encoded_project="$(urlencode "${TOPIC_MANIFEST_REPO}")"
+encoded_file="$(urlencode "${encoded_path}")"
+encoded_ref="$(urlencode "${TOPIC_MANIFEST_SHA}")"
+manifest_url="${server_url}/api/v4/projects/${encoded_project}/repository/files/${encoded_file}/raw?ref=${encoded_ref}"
+
+curl_args=(-fsSL)
+if [[ -n "${CI_JOB_TOKEN:-}" ]]; then
+  curl_args+=(-H "JOB-TOKEN: ${CI_JOB_TOKEN}")
+fi
+
+curl "${curl_args[@]}" "${manifest_url}" -o "${output_path}"
+if grep -qiE '^\s*<!doctype html|^\s*<html' "${output_path}"; then
+  echo "Downloaded topic manifest is HTML, not YAML. Check GitLab API auth for ${TOPIC_MANIFEST_REPO}/${encoded_path}@${TOPIC_MANIFEST_SHA}" >&2
+  exit 1
+fi
+echo "Downloaded topic manifest to ${output_path}"

scripts/generate-pipeline.sh

@@ -46,6 +46,10 @@ find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sort | whi
     - validate-${env}-s3
     - validate-${env}-database"
   fi
+
+  validate_before_script="    - cd \$TG_ROOT
+    - echo \"Working directory:\" && pwd
+    - ls -la"
   
   case "$env" in
     stage) folder_var="YC_STAGE_FOLDER_ID" ;;
@@ -71,6 +75,15 @@ find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sort | whi
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"stage\"'"
   fi
 
+  if [ "$component" = "kafka-topics" ]; then
+    validate_before_script="    - apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && echo \"Working directory:\" && pwd && ls -la"
+    branch_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'"
+  else
+    branch_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
+      when: never
+${branch_rules}"
+  fi
+
   cat >> "$OUTPUT_FILE" << VALIDATE_JOB
 validate-${job_prefix}:
   stage: validate
@@ -82,9 +95,7 @@ ${runner_tags}
     YC_FOLDER_ID: "\$${folder_var}"
     TG_SKIP_DEPENDENCY_OUTPUTS: "true"
   before_script:
-    - cd \$TG_ROOT
-    - echo "Working directory:" && pwd
-    - ls -la
+${validate_before_script}
   script:
     - terragrunt init -upgrade -reconfigure
     - terragrunt validate
@@ -111,6 +122,9 @@ VALIDATE_JOB
     # Для secrets инициализируем зависимости чтобы получить их outputs
     plan_before_script="cd live/${env}/namespace && terragrunt init -upgrade -reconfigure && cd ../s3 && terragrunt init -upgrade -reconfigure && cd ../database && terragrunt init -upgrade -reconfigure && cd ../secrets && ls -la"
   fi
+  if [ "$component" = "kafka-topics" ]; then
+    plan_before_script="apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && ls -la"
+  fi
 
   cat >> "$OUTPUT_FILE" << PLAN_JOB
 plan-${job_prefix}:
@@ -155,13 +169,24 @@ PLAN_JOB
   fi
   
   # Apply rules в зависимости от окружения
-  if [ "$env" = "prod" ]; then
-    apply_rules="    - if: '\$CI_COMMIT_BRANCH == \"master\"'
+  if [ "$component" = "kafka-topics" ]; then
+    apply_before_script="apk add --no-cache bash curl python3 py3-yaml >/dev/null 2>&1 || true; ./scripts/fetch-topic-manifest.sh; ./scripts/validate-kafka-topics.sh; cd \$TG_ROOT && ls -la"
+    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\" && \$TOPIC_APPLY == \"false\"'
+      when: never
+    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\" && \$TOPIC_APPLY != \"false\"'
+      when: on_success
+    - when: never"
+  elif [ "$env" = "prod" ]; then
+    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
+      when: never
+    - if: '\$CI_COMMIT_BRANCH == \"master\"'
       when: ${when_clause}
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"prod\"'
       when: ${when_clause}"
   else
-    apply_rules="    - if: '\$CI_COMMIT_BRANCH == \"stage\"'
+    apply_rules="    - if: '\$TF_COMPONENT == \"kafka-topics\" && \$TOPIC_MANIFEST_PATH != \"\"'
+      when: never
+    - if: '\$CI_COMMIT_BRANCH == \"stage\"'
       when: ${when_clause}
     - if: '\$CI_PIPELINE_SOURCE == \"pipeline\" && \$TARGET_ENV == \"stage\"'
       when: ${when_clause}"
@@ -193,4 +218,4 @@ done
 
 echo "Pipeline generated successfully: $OUTPUT_FILE"
 echo "Found components:"
-find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sed 's|live/||; s|/terragrunt.hcl||' | grep '/' | sort
+find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sed 's|live/||; s|/terragrunt.hcl||' | grep '/' | sort

scripts/validate-kafka-topics.sh

@@ -0,0 +1,131 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+repo_root="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
+env_name="${TOPIC_ENV:-${ENVIRONMENT:-stage}}"
+manifest_path="${TOPIC_MANIFEST_LOCAL_PATH:-${repo_root}/.generated/kafka-topics/${env_name}/manifest.yaml}"
+infra_path="${INFRASTRUCTURE_CONFIG_PATH:-${repo_root}/infrastructure.yaml}"
+
+python3 - "${manifest_path}" "${infra_path}" "${env_name}" <<'PY'
+import os
+import sys
+
+try:
+    import yaml
+except Exception as exc:
+    raise SystemExit(f"pyyaml is required for Kafka topics validation: {exc}")
+
+manifest_path, infra_path, env = sys.argv[1:4]
+
+def load_yaml(path):
+    if not os.path.exists(path):
+        raise SystemExit(f"File not found: {path}")
+    with open(path, "r", encoding="utf-8") as fh:
+        return yaml.safe_load(fh) or {}
+
+def env_value(value, env, default=None):
+    if value is None:
+        return default
+    if isinstance(value, dict):
+        if env in value:
+            return value[env]
+        if "_default" in value:
+            return value["_default"]
+        return default
+    return value
+
+manifest = load_yaml(manifest_path)
+infra = load_yaml(infra_path)
+
+env_cfg = (infra.get("environments") or {}).get(env)
+if not isinstance(env_cfg, dict):
+    raise SystemExit(f"Environment '{env}' is not defined in {infra_path}")
+
+cluster_refs = env_cfg.get("kafka_cluster_refs") or {}
+policy = env_cfg.get("kafka_policy") or {}
+topics = ((manifest.get("kafka") or {}).get("topics"))
+
+if not isinstance(topics, list):
+    raise SystemExit("kafka.topics must be an array")
+
+allowed_cleanup = {"delete", "compact"}
+allowed_deletion = {"orphan", "delete"}
+errors = []
+
+for idx, topic in enumerate(topics):
+    prefix = f"kafka.topics[{idx}]"
+    if not isinstance(topic, dict):
+        errors.append(f"{prefix}: must be an object")
+        continue
+
+    name = topic.get("name")
+    owner = topic.get("owner")
+    cluster_ref = topic.get("clusterRef")
+
+    if not name:
+        errors.append(f"{prefix}.name is required")
+    if not owner:
+        errors.append(f"{prefix}.owner is required")
+    if not cluster_ref:
+        errors.append(f"{prefix}.clusterRef is required")
+        continue
+    if cluster_ref not in cluster_refs:
+        errors.append(f"{prefix}.clusterRef '{cluster_ref}' is not defined in infrastructure.yaml for env '{env}'")
+        continue
+
+    cluster = cluster_refs[cluster_ref] or {}
+    partitions = env_value(topic.get("partitions"), env, cluster.get("default_partitions"))
+    replication = env_value(topic.get("replicationFactor"), env, cluster.get("default_replication_factor"))
+
+    if partitions is None:
+        errors.append(f"{prefix}.partitions is required or cluster default_partitions must be set")
+    if replication is None:
+        errors.append(f"{prefix}.replicationFactor is required or cluster default_replication_factor must be set")
+
+    try:
+        partitions_num = int(partitions)
+        if partitions_num < 1:
+            errors.append(f"{prefix}.partitions must be >= 1")
+    except Exception:
+        errors.append(f"{prefix}.partitions must be a number")
+
+    try:
+        replication_num = int(replication)
+        if replication_num < 1:
+            errors.append(f"{prefix}.replicationFactor must be >= 1")
+    except Exception:
+        replication_num = None
+        errors.append(f"{prefix}.replicationFactor must be a number")
+
+    max_replication = cluster.get("max_replication_factor")
+    if replication_num is not None and max_replication is not None and replication_num > int(max_replication):
+        errors.append(f"{prefix}.replicationFactor={replication_num} exceeds cluster max_replication_factor={max_replication}")
+
+    deletion_policy = topic.get("deletionPolicy", "orphan")
+    if deletion_policy not in allowed_deletion:
+        errors.append(f"{prefix}.deletionPolicy must be one of {sorted(allowed_deletion)}")
+    if deletion_policy == "delete" and not bool(policy.get("allow_delete", False)):
+        errors.append(f"{prefix}.deletionPolicy=delete is forbidden by kafka_policy.allow_delete=false")
+
+    merged_config = {}
+    merged_config.update(cluster.get("default_topic_config") or {})
+    merged_config.update(topic.get("config") or {})
+
+    cleanup_policy = env_value(merged_config.get("cleanup.policy"), env)
+    if cleanup_policy is not None and cleanup_policy not in allowed_cleanup:
+        errors.append(f"{prefix}.config.cleanup.policy must be one of {sorted(allowed_cleanup)}")
+
+    min_isr = env_value(merged_config.get("min.insync.replicas"), env)
+    if min_isr is not None and replication_num is not None:
+        try:
+            min_isr_num = int(min_isr)
+            if min_isr_num > replication_num:
+                errors.append(f"{prefix}.config.min.insync.replicas={min_isr_num} is greater than replicationFactor={replication_num}")
+        except Exception:
+            errors.append(f"{prefix}.config.min.insync.replicas must be a number")
+
+if errors:
+    raise SystemExit("Kafka topics validation failed:\n- " + "\n- ".join(errors))
+
+print(f"Kafka topics manifest is valid: {manifest_path}")
+PY