streamtime_kafka_replicator (Resource)

Copies topics and consumer groups from one Kafka cluster to another. Creating a replicator deploys it and waits until it is ready. Changes to its settings are saved, but a running replicator keeps its current settings until Streamtime can apply changes to running replicators.

Example Usage 

# Copy topics from one Streamtime Kafka cluster to another. Apply waits until the replicator is ready.
resource "streamtime_kafka_replicator" "mirror" {
  identifier           = "east-to-west"
  replicator_provider  = "MirrorMaker2"
  source_kafka_id      = streamtime_kafka.east.identifier
  destination_kafka_id = streamtime_kafka.west.identifier

  # The credentials for both clusters are stored as secrets that must already exist.
  source_secret_path      = "builtin:///replicators/east-to-west/source-secret"
  destination_secret_path = "builtin:///replicators/east-to-west/destination-secret"

  replication_mode         = "active-passive"
  lag_alert_threshold      = 1000
  lag_alert_threshold_unit = "messages"

  topic_whitelist = "orders.*,payments"

  timeouts = {
    create = "60m"
    delete = "60m"
  }
}

# To copy from a cluster outside Streamtime, leave out source_kafka_id and describe how to reach it:
#   source_connection_details = jsonencode({
#     bootstrap_servers = "kafka.example.com:9093"
#     security_protocol = "SASL_SSL"
#     sasl_mechanism    = "PLAIN"
#   })

Schema

Required

  • destination_secret_path (String) Where the credentials for the destination cluster are stored, as a secret URI. The secret must already exist.
  • identifier (String) Unique name: up to 24 lowercase letters, digits and dashes. Changing it creates a new replicator.
  • lag_alert_threshold (Number) Replication lag that raises an alert, in the unit of lag_alert_threshold_unit.
  • lag_alert_threshold_unit (String) Unit of the lag threshold: messages, bytes or seconds.
  • replication_mode (String) active-passive or active-active.
  • replicator_provider (String) How data is copied: MirrorMaker2 or Cluster Link. Changing it creates a new replicator.
  • source_secret_path (String) Where the credentials for the source cluster are stored, as a secret URI, for example builtin:///replicators//source-secret. The secret must already exist.

Optional

  • cluster_link_advanced_settings (String) Cluster Link settings, as a JSON object set with jsonencode().
  • destination_connection_details (String) How to reach an external destination cluster, as a JSON object set with jsonencode(), with the same fields as source_connection_details.
  • destination_kafka_id (String) Identifier of the Streamtime Kafka cluster to copy to. Leave it out for an external cluster and set destination_connection_details.
  • groups_blacklist (String) Consumer groups not to copy, as a comma-separated list or a pattern.
  • groups_whitelist (String) Consumer groups to copy, as a comma-separated list or a pattern.
  • link_mode (String) Bidirectional, Destination or Source. Required when source_initiated_link is true.
  • mirrormaker2_advanced_settings (String) MirrorMaker2 settings, as a JSON object set with jsonencode().
  • source_connection_details (String) How to reach an external source cluster, as a JSON object set with jsonencode(): bootstrap_servers, security_protocol, and for SASL sasl_mechanism (and cluster_id for Cluster Link).
  • source_initiated_link (Boolean) Whether the link is started from the source cluster (Cluster Link only). Requires link_mode.
  • source_kafka_id (String) Identifier of the Streamtime Kafka cluster to copy from. Leave it out for an external cluster and set source_connection_details.
  • timeouts (Attributes) (see below for nested schema)
  • topic_blacklist (String) Topics not to copy, as a comma-separated list or a pattern.
  • topic_whitelist (String) Topics to copy, as a comma-separated list or a pattern.

Read-Only

  • configuration_state (String) Whether the settings are complete: incomplete, valid or invalid.
  • fleet (String) Identifier of the fleet the replicator runs on, chosen from its clusters.
  • generation (Number) Version of the settings; it goes up when they change.
  • id (String) Same as identifier. Use the identifier to import.
  • observed_generation (Number) Version of the settings that is running.
  • reconciliation_state (String) Progress of the last change: pending, queued, running, succeeded or failed.
  • status (String) Health of the replicator, for example Provisioning, Healthy or Error.
  • warnings (String) Problems Streamtime found with where the replicator runs, as JSON. Read it with jsondecode().

Nested Schema for timeouts

Optional:

  • create (String) A string that can be parsed as a duration consisting of numbers and unit suffixes, such as “30s” or “2h45m”. Valid time units are “s” (seconds), “m” (minutes), “h” (hours).
  • delete (String) A string that can be parsed as a duration consisting of numbers and unit suffixes, such as “30s” or “2h45m”. Valid time units are “s” (seconds), “m” (minutes), “h” (hours). Setting a timeout for a Delete operation is only applicable if changes are saved into state before the destroy operation occurs.

Import

Import is supported using the following syntax:

The terraform import command can be used, for example:

# Import by identifier.
terraform import streamtime_kafka_replicator.mirror east-to-west