diff --git a/.gitignore b/.gitignore index b21fa44..95c5d3e 100644 --- a/.gitignore +++ b/.gitignore @@ -20,6 +20,7 @@ terragrunt.hcl.backup # Generated files .gitlab-ci.generated.yml +.generated/ # IDE .idea/ diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 6d1d849..9a8c375 100644 --- a/.gitlab-ci.yml +++ b/.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\/.*/' -## \ No newline at end of file +## diff --git a/infrastructure.yaml b/infrastructure.yaml index db21e84..7f2160b 100644 --- a/infrastructure.yaml +++ b/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: diff --git a/live/prod/kafka-topics/terragrunt.hcl b/live/prod/kafka-topics/terragrunt.hcl new file mode 100644 index 0000000..92b7d2b --- /dev/null +++ b/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", "") +} diff --git a/live/terragrunt.hcl b/live/terragrunt.hcl index fe9ee37..b7a3674 100644 --- a/live/terragrunt.hcl +++ b/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" diff --git a/modules/kafka-topics-yc/main.tf b/modules/kafka-topics-yc/main.tf new file mode 100644 index 0000000..7021344 --- /dev/null +++ b/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] +} diff --git a/modules/kafka-topics-yc/outputs.tf b/modules/kafka-topics-yc/outputs.tf new file mode 100644 index 0000000..d43018b --- /dev/null +++ b/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] +} diff --git a/modules/kafka-topics-yc/variables.tf b/modules/kafka-topics-yc/variables.tf new file mode 100644 index 0000000..3662664 --- /dev/null +++ b/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 +} diff --git a/modules/kafka-topics-yc/versions.tf b/modules/kafka-topics-yc/versions.tf new file mode 100644 index 0000000..fd126d1 --- /dev/null +++ b/modules/kafka-topics-yc/versions.tf @@ -0,0 +1,3 @@ +terraform { + required_version = ">= 1.5.0" +} diff --git a/scripts/fetch-topic-manifest.sh b/scripts/fetch-topic-manifest.sh new file mode 100755 index 0000000..426563a --- /dev/null +++ b/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*&2 + exit 1 +fi +echo "Downloaded topic manifest to ${output_path}" diff --git a/scripts/generate-pipeline.sh b/scripts/generate-pipeline.sh index c4dcc1b..01b1e91 100755 --- a/scripts/generate-pipeline.sh +++ b/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 \ No newline at end of file +find live -name "terragrunt.hcl" -not -path "*/.terragrunt-cache/*" | sed 's|live/||; s|/terragrunt.hcl||' | grep '/' | sort diff --git a/scripts/validate-kafka-topics.sh b/scripts/validate-kafka-topics.sh new file mode 100755 index 0000000..d1d3ffe --- /dev/null +++ b/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