Distributor Helm Readme

helm registry login -u [your-user] registry.axual.io/axual-charts
helm upgrade --install local-distributor oci://registry.axual.io/axual-charts/distributor --version 5.5.2 -f values.yaml

global:
  # Globally override the registry to pull images from.
  imageRegistry: "registry.axual.io"
  # Globally override the list of ImagePullSecrets provided, used for init containers
  imagePullSecrets: []

# Configure the Strimzi Cluster Operator version installed. Used to determine the correct image tag and default logging to use
strimzi:
  version: "0.48.0"

connect:
  # Contains the image to use for connect
  image:
    repository: "axual/distributor"
    # Specify the exact tag to override version deduction from strimzi.version value
    # tag: "5.5.2-0.48.0"

  # The number of connect replicas to spin up
  replicas: 3

  # Should connector resources, or Connector CRDs be used, needs to be true
  useConnectorResources: true

  # Use Strimzi Rack Awareness
  rack:
    enabled: true
    topologyKey: topology.kubernetes.io/zone

  #The bootstrap url used to connect to the kafka cluster
  bootstrapServers: "cluster01-kafka-bootstrap:9093"
  # The consumer group identifier used by this Kafka Connect cluster
  groupId: "_rbsoft-multi-distributor-group"
  # Contains the internal connect topics settings
  topics:
    # Set this to the replication factor to use on the source Kafka cluster
    replicationFactor: 1
    config:
      # Name of the Kafka Connect config topic
      name: "_axual-dta-distributor-config"
    status:
      # Name of the Kafka Connect status topic
      name: "_axual-dta-distributor-status"
      # Set the number of partitions used for the status topic
      partitions: 2
    offset:
      # Name of the Kafka Connect offset topic
      name: "_axual-dta-distributor-offset"
      # Set the number of partitions used for the offset topic
      partitions: 25

  # Control the standard way Kafka Connect discovers available plugins
  # Available modes:
  # - ONLY_SCAN: Legacy behavior, scans all classes (slowest but most compatible) (Default for Distributor)
  # - HYBRID_WARN: Uses both methods, warns about plugins missing ServiceLoader manifests
  # - HYBRID_FAIL: Like HYBRID_WARN but fails on missing manifests (used in test environments)
  # - SERVICE_LOAD: Uses only ServiceLoader (fastest but requires all plugins to have manifests)
  # Note: All modes introduced in Kafka 3.6.0. Prior versions only support classpath scanning.
  pluginDiscovery: SERVICE_LOAD

  # Set the rest of the Kafka Connect configuration
  config:
    key.converter: JsonConverter
    value.converter: JsonConverter

    # Use this if you need SASL authentication for the source cluster
    #  sasl:
    #      # Set to true to enable a SASL connection
    #     enabled: false
    #     type: "PLAIN" # only PLAIN and SCRAM-SHA-512 are supported
    #     username: ""
    #     password: ""

    # Contains the TLS settings for connecting to the kafka cluster
    tls:
      enabled: true
      createCaCertsSecret: false
      # if createTruststoreCaSecret is true, set the CA certs below
      #    caCerts:
      #      one_ca.crt: |
      #        -----BEGIN CERTIFICATE-----
      #      other_ca.crt: |
      #        -----BEGIN CERTIFICATE-----

      # if createTruststoreCaSecret is false, the caCerts need to be set
      # with an existing secret (name) and the name of the cert inside the
      # secret
      # caSecret:
      #  secretName: your_custom_ca_secret
      #  keyForCertificate: your_custom_ca_cert
      caSecret:
        secretName: "cluster01-cluster-ca-cert"
        keyForCertificate: "ca.crt"

      # Configure authentication using a client certificate
      authentication:
        enabled: true
        createTlsClientSecret: false
        #           clientCert: |
        #              -----BEGIN CERTIFICATE-----
        #              -----END CERTIFICATE-----
        #           clientKey: |
        #              -----BEGIN PRIVATE KEY-----
        #              -----END PRIVATE KEY-----

        # if a TLS secret already exists with the client credentials, provide the name here
        secretName: "axual-cluster01-distributor"

    # Set logging rules, the different types and formats can be found in Strimzi documentation
    # If logging is not set, a default version will be deduced from the value of strimzi.version
    logging:
      type: inline
      loggers:
        rootLogger.level: "INFO"
        logger.distributorCommon.name: "io.axual.distributor.common"
        logger.distributorCommon.level: "INFO"
        logger.distributorMessage.name: "io.axual.distributor.message"
        logger.distributorMessage.level: "INFO"
        logger.distributorOffset.name: "io.axual.distributor.offset"
        logger.distributorOffset.level: "INFO"
        logger.distributorSchema.name: "io.axual.distributor.schema"
        logger.distributorSchema.level: "INFO"
        logger.kafkaConsumer.name: "org.apache.kafka.clients.consumer"
        logger.kafkaConsumer.level: "WARN"
        logger.kafkaProducer.name: "org.apache.kafka.clients.producer"
        logger.kafkaProducer.level: "WARN"
        logger.kafkaAdmin.name: "org.apache.kafka.clients.admin"
        logger.kafkaAdmin.level: "WARN"

# Used for creating the required Kafka ACLs and topics on the Kafka cluster
init:
 enabled: true
 # A list of Kafka principal names that should be used in the ACLS
 principals:
  - "User:CN=rbsoft distributor client for multi instance,OU=Development,O=Axual,L=Utrecht,ST=Utrecht,C=NL"

 # Secrets containing the TLS certificates for applying topics and acl to kafka
 tls:
  # -- Existing Keypair secret name
  keypairSecretName: "axual-remote-2-super-user-cert"
  # -- Existing Keypair key name
  keypairSecretKeyName: "tls.key"
  # -- Existing Keypair certificate name
  keypairSecretCertName: "tls.crt"
  # -- Existing Truststore secret name
  truststoreCaSecretName: "cluster01-cluster-ca-cert"
  # -- Existing Truststore certificate name
  truststoreCaSecretCertName: "ca.crt"

# Contains the distribution settings
distribution:
  enabled: true

  # Axual registration data
  tenantShortName: "axual"
  instanceShortName: "dta"

  # Prefix values for topic and group ACLs
  prefixAclSettings:
    topicPrefix: "rbsoft-dual-"
    groupPrefix: "rbsoft-dual-"

  sourceCluster:
    name: cluster01
    # Contains instance topic information for the cluster
    topics:
      replicationFactor: 1
      # Timestamps topic for offset distribution
      timestamps:
        # Use name override to not use default name of _<tenant>-<instance>-consumer-timestamps
        nameOverride: ""
        partitions: 25
      # Schema topic for schema distribution
      schemas:
        # Use name override to not use default name of _<tenant>-<instance>-schemas
        nameOverride: ""

    topicPattern: '{tenant}-{instance}-{environment}-{topic}'
    groupPattern: '{tenant}-{instance}-{environment}-{group.id}'
    patternDefaultValues:
      environment: test3

    # Connection information, useful for connectors that use source as target
    bootstrapServers: "cluster01-kafka-bootstrap:9093"
    securityProtocol: "SSL"
    additionalClientConfigs:
      ssl.enabled.protocols: TLSv1.2

    # Use this if the connection to the source cluster should use SASL
    #sasl:
    #  mechanism: SCRAM-SHA-512
    #  username: my-user
    #  password: my-password
    tls:
      # CA Certificates that might be used in this cluster
      caCerts:
        - |
          -----BEGIN CERTIFICATE-----
          -----END CERTIFICATE-----
        - |
          -----BEGIN CERTIFICATE-----
          -----END CERTIFICATE-----
      clientCert: |
        -----BEGIN CERTIFICATE-----
        -----END CERTIFICATE-----
      clientKey: |
        -----BEGIN PRIVATE KEY-----
        -----END PRIVATE KEY-----
      # No password, unencrypted clientKey
      #clientKeyPassword: ""

  # Information about target clusters
  clusters:
    # information about cluster 02
    cluster02:
      name: cluster02
      # Contains instance topic information for the cluster
      topics:
        replicationFactor: 1
        timestamps:
          nameOverride: ""
          partitions: 25
        schemas:
          nameOverride: ""

      topicPattern: '{tenant}-{instance}-{environment}-{topic}'
      groupPattern: '{tenant}-{instance}-{environment}-{group.id}'
      patternDefaultValues:
        environment: test3

      bootstrapServers: "cluster02-kafka-bootstrap:9093"
      securityProtocol: "SSL"
      additionalClientConfigs:
        ssl.enabled.protocols: TLSv1.2

      tls:
        caCerts:
          - |
            -----BEGIN CERTIFICATE-----
            -----END CERTIFICATE-----
          - |
            -----BEGIN CERTIFICATE-----
            -----END CERTIFICATE-----
        clientCert: |
          -----BEGIN CERTIFICATE-----
          -----END CERTIFICATE-----
        clientKey: |
          -----BEGIN PRIVATE KEY-----
          -----END PRIVATE KEY-----
        #clientKeyPassword: ""

    # information about cluster 03
    cluster03:
      name: cluster03
      topics:
        replicationFactor: 1
        timestamps:
          nameOverride: ""
          partitions: 25
        schemas:
          nameOverride: ""

      topicPattern: '{tenant}-{instance}-{environment}-{topic}'
      groupPattern: '{tenant}-{instance}-{environment}-{group.id}'
      patternDefaultValues:
        environment: test3

      bootstrapServers: "cluster03-kafka-bootstrap:9093"
      securityProtocol: "SSL"
      additionalClientConfigs:
        ssl.enabled.protocols: TLSv1.2

      tls:
        caCerts:
          - |
            -----BEGIN CERTIFICATE-----
            -----END CERTIFICATE-----
          - |
            -----BEGIN CERTIFICATE-----
            -----END CERTIFICATE-----
        clientCert: |
          -----BEGIN CERTIFICATE-----
          -----END CERTIFICATE-----
        clientKey: |
          -----BEGIN PRIVATE KEY-----
          -----END PRIVATE KEY-----
        #clientKeyPassword: ""

  # Settings for the Offset Committer connector
  offsetCommitter:
    enabled: true
    # Determine the number of connect tasks to run for the offset committer
    maxTasks: 1
    # Use the override to use a custom consumer group for this connector
    #groupIdOverride: ""

    # Do not perform a connectivity check on starting the connector
    skipConnectionValidation: true

    # The amount of time subtracted from the received timestamp to ensure the offset commit points to a record that was distributed before the timestamp
    distributionOffsetMs: 15000
    # Set the number of items in the topic and group resolve cache
    topicCacheSize: 200
    groupCacheSize: 200
    # Reinitialisation trigger settings for the internal senders when continuous errors occur
    reinitializeOnContinuousError:
      maximumCount: 500
      maximumTimeMs: 10000

    # Any additional client configs needed to connect to the source cluster to commit the offsets
    additionalClientConfigs:
      ssl.enabled.protocols: TLSv1.2,TLSv1.3
      ssl.endpoint.identification.algorithm: ''

    # Override the default Kafka Connect Sink Task Consumer settings
    consumerOverrides:
      max.poll.size: 5

  # Settings for the Schema Distributor connector
  schemaDistributor:
    enabled: false
    # Schemas are distributed one way to the following clusters
    targetClusterNames:
      - "cluster02"
      - "cluster03"
    additionalClientConfigs:
      acks: all
      retries: 0
      max.in.flight.requests.per.connection: 1
      enable.idempotence: false
      ssl.enabled.protocols: TLSv1.2,TLSv1.3
      ssl.endpoint.identification.algorithm: ''
    consumerOverrides:
      max.poll.size: 10

  # Axual uses a multi level distribution model to make sure records and offsets arrive at the correct cluster
  # This is a dictionary containing the distribution levels relevant to the source cluster.
  # The level number is 1 to 32
  levels:
    1:
      targetClusterName: "cluster02"

      messageDistributor:
        enabled: true
        maxTasks: 2
        groupIdOverride: ""
        topicRegex: 'axual-dta-.*'
        topicCacheSize: 1000
        additionalTargetClientConfigs: {}
        consumerOverrides:
          max.poll.size: 8

      offsetDistributor:
        enabled: true
        maxTasks: 2
        groupIdOverride: ""
        windowSizeMs: 15000
        topicCacheSize: 1000
        groupCacheSize: 1000
        additionalLocalClientConfigs: {}
        additionalTargetClientConfigs: {}
        consumerOverrides:
          max.poll.size: 6

    2:
      targetClusterName: "cluster03"

      messageDistributor:
        enabled: true
        maxTasks: 2
        groupIdOverride: ""
        topicRegex: 'axual-dta-.*'
        topicCacheSize: 1000
        additionalTargetClientConfigs: {}
        consumerOverrides:
          max.poll.size: 9

      offsetDistributor:
        enabled: true
        maxTasks: 2
        groupIdOverride: ""
        windowSizeMs: 15000
        topicCacheSize: 1000
        groupCacheSize: 1000
        additionalLocalClientConfigs: {}
        additionalTargetClientConfigs: {}
        consumerOverrides:
          max.poll.size: 7