connectivity.conf

ditto {
  service-name = "connectivity"

  mongodb {
    database = "connectivity"
    database = ${?MONGO_DB_DATABASE}
  }

  persistence.operations.delay-after-persistence-actor-shutdown = 5s
  persistence.operations.delay-after-persistence-actor-shutdown = ${?DELAY_AFTER_PERSISTENCE_ACTOR_SHUTDOWN}

  signal-enrichment {
    # Which caching signal enrichment facade is used by the signal-enrichment.provider
    caching-signal-enrichment-facade.provider = ${?CONNECTIVITY_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
  }

  connectivity {

    user-indicated-errors-base = [
      # Kafka
      {exceptionName: "org.apache.kafka.common.errors.SaslAuthenticationException", messagePattern: ".*"}
      {exceptionName: "org.apache.kafka.common.errors.UnsupportedSaslMechanismException", messagePattern: ".*"}
      {exceptionName: "org.apache.kafka.common.errors.IllegalSaslStateException", messagePattern: ".*"}
      {exceptionName: "org.apache.kafka.common.errors.SslAuthenticationException", messagePattern: ".*"}
      {exceptionName: "org.apache.kafka.common.errors.TimeoutException", messagePattern: ".*"}
      # MQTT
      {exceptionName: "com.hivemq.client.mqtt.exceptions.ConnectionClosedException", messagePattern: "Server closed connection without DISCONNECT.*"}
      {exceptionName: "com.hivemq.client.mqtt.exceptions.ConnectionClosedException", messagePattern: "io.netty.channel.unix.Errors$NativeIoException.*"}
      # MQTT 3.1.1 - treat error codes from the broker as user indicated errors
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3ConnAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3DisconnectException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3PubAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3PubRecException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3SubAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt3.exceptions.Mqtt3UnsubAckException", messagePattern: ".*"}
      # MQTT 5 - treat error codes from the broker as user indicated errors
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5AuthException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5DisconnectException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5PubAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5PubRecException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5SubAckException", messagePattern: ".*"}
      {exceptionName: "com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5UnsubAckException", messagePattern: ".*"}
      # RabbitMQ
      {exceptionName: "com.rabbitmq.client.AuthenticationFailureException", messagePattern: ".*"}
      {exceptionName: "com.rabbitmq.client.ShutdownSignalException", messagePattern: ".*"}
      {exceptionName: "javax.net.ssl.SSLException", messagePattern: ".*"}
      {exceptionName: "java.net.SocketException", messagePattern: "Socket closed.*"}
      # SSH Tunnel
      {exceptionName: "java.nio.channels.UnresolvedAddressException", messagePattern: ".*"}
      {exceptionName: "org.apache.sshd.common.SshException", messagePattern: ".*authentication.*"}
      {exceptionName: "java.net.ConnectException", messagePattern: ".*Connection refused.*"}
      {exceptionName: "java.lang.RuntimeException", messagePattern: ".*load key from.*"} //thrown when private key could not be loaded
      # AMQP 1.0
      {exceptionName: "org.apache.qpid.jms.provider.ProviderException", messagePattern: ".*"}
      {exceptionName: "org.apache.qpid.jms.provider.exceptions.ProviderSecurityException", messagePattern: ".*"}
      {exceptionName: "org.apache.qpid.jms.provider.exceptions.ProviderInvalidDestinationException", messagePattern: ".*"}
      # HTTP
      {exceptionName: "org.eclipse.ditto.jwt.model.JwtInvalidException", messagePattern: ".*"}
      {exceptionName: "akka.http.impl.engine.client.OutgoingConnectionBlueprint.UnexpectedConnectionClosureException", messagePattern: ".*"}
      {exceptionName: "akka.stream.StreamTcpException", messagePattern: ".*Connection reset by peer.*"}
      # Common cert exceptions
      {exceptionName: "java.security.cert.CertificateException", messagePattern: ".*"}
      {exceptionName: "java.security.cert.CertPathValidatorException", messagePattern: ".*"}
      {exceptionName: "sun.security.provider.certpath.SunCertPathBuilderException", messagePattern: ".*"}
    ]

    user-indicated-errors-extended = []
    user-indicated-errors-extended = ${?USER_INDICATED_ERRORS}

    user-indicated-errors = ${ditto.connectivity.user-indicated-errors-base} ${ditto.connectivity.user-indicated-errors-extended}

    connection {
      # A comma separated string of hostnames to which http requests will allowed. This overrides the blocked
      # hostnames i.e if a host is blocked *and* allowed, it will be allowed.
      allowed-hostnames = ""
      #allowed-hostnames = "localhost"
      allowed-hostnames = ${?CONNECTIVITY_CONNECTION_ALLOWED_HOSTNAMES}

      # A comma separated string of blocked hostnames to which no http requests will be sent out.
      # Set it to an empty string to permit outgoing HTTP connections against private IP addresses.
      blocked-hostnames = ""
      #blocked-hostnames = "localhost"
      blocked-hostnames = ${?CONNECTIVITY_CONNECTION_BLOCKED_HOSTNAMES}

      # A comma separated string of blocked subnets to which no http requests will be sent out.
      # Specify subnets to block in CIDR format e.g. "11.1.0.0/16"
      blocked-subnets = ""
      blocked-subnets = ${?CONNECTIVITY_CONNECTION_BLOCKED_SUBNETS}

      # A regex which is used to determine blocked hostnames.
      blocked-host-regex = ""
      blocked-host-regex = ${?CONNECTIVITY_CONNECTION_BLOCKED_HOST_REGEX}

      # Number of sources per connection
      max-source-number = 4
      max-source-number = ${?CONNECTION_SOURCE_NUMBER}
      # Number of targets per connection
      max-target-number = 4
      max-target-number = ${?CONNECTION_TARGET_NUMBER}

      supervisor {
        exponential-backoff {
          min = 1s
          max = 120s
          random-factor = 1.0
          corrupted-receive-timeout = 600s
        }
      }

      snapshot {
        threshold = 10
        interval = 15m
      }

      activity-check {
        # the interval of how long to keep a closed, "inactive" Connection in memory
        inactive-interval = 45m
        inactive-interval = ${?CONNECTION_ACTIVITY_CHECK_INTERVAL} # may be overridden with this environment variable

        # the interval of how long to keep a deleted Connection in memory:
        deleted-interval = 5m
        deleted-interval = ${?CONNECTION_ACTIVITY_CHECK_DELETED_INTERVAL}
      }

      # how long for connection actor to wait for response from client actors
      # by default this value is very high because connection establishment can take very long and if we timeout too
      # early the connection is not subscribed for events properly
      # this timeout needs to be smaller then ditto.gateway.http.request-timeout in gateway.conf
      client-actor-ask-timeout = 55s
      client-actor-ask-timeout = ${?CONNECTIVITY_CLIENT_ACTOR_ASK_TIMEOUT}

      # how often the connection actor will retry starting a failing client actor. If exceeded, errors will be
      # escalated to the supervisor, which will effectively cause the whole connection to be restarted.
      client-actor-restarts-before-escalation = 3
      client-actor-restarts-before-escalation = ${?CONNECTIVITY_CLIENT_ACTOR_RETRIES_BEFORE_ESCALATION}

      # Whether to start 1 client actor on each node or to start all client actors on the same node.
      # It is not possible to start multiple client actors on multiple nodes.
      all-client-actors-on-one-node = false
      all-client-actors-on-one-node = ${?CONNECTIVITY_ALL_CLIENT_ACTORS_ON_ONE_NODE}

      # Whether double decoding of usernames and passwords in connection URIs is enabled.
      double-decoding-enabled = true
      double-decoding-enabled = ${?CONNECTIVITY_DOUBLE_DECODING_ENABLED}

      acknowledgement {
        # lifetime of ack forwarder. Must be bigger than the largest possible command timeout (60s)
        forwarder-fallback-timeout = 65s

        # lifetime of collector of source acknowledgements. usually terminates before this long when all acks arrive.
        collector-fallback-lifetime = 100s

        # how long to wait on the collector before acknowledging negatively on the transport layer.
        collector-fallback-ask-timeout = 120s

        # size budget for the payload of issued acknowledgements by a connection target.
        issued-max-bytes = 100000
        issued-max-bytes = ${?CONNECTIVITY_ACKNOWLEDGEMENT_ISSUED_MAX_BYTES}
      }

      # how often to attempt acknowledgement label declaration for as long as it is not successful
      ack-label-declare-interval = 10s
      ack-label-declare-interval = ${?CONNECTIVITY_ACK_LABEL_DECLARE_INTERVAL}

      priority-update-interval = 24h
      priority-update-interval = ${?CONNECTIVITY_PRIORITY_UPDATE_INTERVAL}

      amqp10 {
        consumer {
          # When to forget an NAcked redelivery=true message -- those may be consumed by another consumer
          # Equal to the max time interval per redelivery per message during subscriber downtime
          redelivery-expectation-timeout = 120s
          redelivery-expectation-timeout = ${?AMQP10_CONSUMER_REDELIVERY_EXPECTATION_TIMEOUT}

          throttling {
            enabled = false
            enabled = ${?AMQP10_CONSUMER_THROTTLING_ENABLED}
            enabled = ${?AMQP10_CONSUMER_RATE_LIMIT_ENABLED} # this is an alias which previously was used

            # Interval at which the consumer is throttled. Must be > 0s.
            interval = 1s
            interval = ${?AMQP10_CONSUMER_THROTTLING_INTERVAL}

            # The maximum number of messages the consumer is allowed to receive within the configured
            # throttling interval e.g. 100 msgs/s. Must be > 0.
            limit = 100
            limit = ${?AMQP10_CONSUMER_THROTTLING_LIMIT}

            # Throttling detection tolerance in percent e.g. for a tolerance of 5% if consumed messages are above
            # limit - 5% per interval the source is considered throttled.
            throttling-detection-tolerance = 0.05
            throttling-detection-tolerance = ${?AMQP_10_CONSUMER_THROTTLING_DETECTION_TOLERANCE}

            # How many unacknowledged messages are allowed at any time as factor of ${limit} - must be >= 1.0
            # This limit couples latency with throughput (long latency before ack -> lower throughput)
            max-in-flight-factor = 2.0
            max-in-flight-factor = ${?AMQP10_CONSUMER_THROTTLING_MAX_IN_FLIGHT_FACTOR}
          }
        }

        publisher {
          # If a message can't be published it is put in a queue. Further messages are dropped when the queue is full.
          max-queue-size = 1000
          max-queue-size = ${?AMQP10_PUBLISHER_MAX_QUEUE_SIZE}

          # Messages to publish in parallel per AMQP-Publisher (one per connectivity client)
          # jms-connection-handling-dispatcher will be used.
          parallelism = 10
          parallelism = ${?AMQP10_PUBLISHER_PARALLELISM}
        }

        // How many producers to cache per client actor (in addition to static addresses).
        // If 0 or negative, no message can be sent to any reply-to address or addresses containing placeholders that
        // do not match any target address.
        producer-cache-size = 10
        producer-cache-size = ${?AMQP10_PRODUCER_CACHE_SIZE}

        global-connect-timeout = 15s
        global-connect-timeout = ${?AMQP10_GLOBAL_CONNECT_TIMEOUT}

        global-send-timeout = 60s
        global-send-timeout = ${?AMQP10_GLOBAL_SEND_TIMEOUT}

        global-request-timeout = 5s
        global-request-timeout = ${?AMQP10_GLOBAL_REQUEST_TIMEOUT}

        global-prefetch-policy-all-count = 10
        global-prefetch-policy-all-count = ${?AMQP10_GLOBAL_PREFETCH_POLICY_ALL_COUNT}

        # Configuration for backoff of the producers and consumers. This will be used when the creation of a consumer
        # or producer fails with an error message of the JMS client (can happen when e.g. user is not authorized
        # on a source/target).
        # As the actor will instantly try to re-establish the sours/targets, this may lead to a lot of error logs.
        # Thats why this backoff configuration is used. On the first error the actor will wait min-timeout before
        # retrying to establish the source/target, afterwards it will increase its waiting time step by step until
        # max-timeout.
        backoff.timeout {
          # initial / minimum timeout when backing off
          min-timeout = 1s
          min-timeout = ${?AMQP10_BACK_OFF_MIN_TIMEOUT}
          # maximum backoff timeout
          max-timeout = 10m
          max-timeout = ${?AMQP10_BACK_OFF_MAX_TIMEOUT}
        }

        hmac-algorithms {
          az-sasl = "org.eclipse.ditto.connectivity.service.messaging.signing.AzSaslSigningFactory"
        }

        initial-consumer-resource-status-ask-timeout = 500ms
        initial-consumer-resource-status-ask-timeout = ${?AMQP10_INITIAL_CONSUMER_RESOURCE_STATUS_ASK_TIMEOUT}
      }

      amqp091 {
        publisher {
          # Lifetime of an entry in the cache 'outstandingAcks'.
          # It is an upper bound for the timeout of any command requesting acknowledgements from this publisher.
          # Ideally between the maximum timeout (60s) and the acknowledgement forwarder lifetime (100s).
          # No other publisher actor requires a cache TTL config because their clients take care of message ID tracking.
          pending-ack-ttl = 1m
        }
      }

      mqtt {
        # maximum mumber of MQTT messages to buffer in a source (presumably for at-least-once and exactly-once delivery)
        source-buffer-size = 8
        source-buffer-size = ${?CONNECTIVITY_MQTT_SOURCE_BUFFER_SIZE}

        # If a message can't be published it is put in a queue. Further messages are dropped when the queue is full.
        max-queue-size = 1000
        max-queue-size = ${?CONNECTIVITY_MQTT_MAX_QUEUE_SIZE}

        # The number of threads to use for the underlying event loop of the MQTT client.
        #  When configured to "0", the size is determined based on "the available processor cores * 2".
        event-loop-threads = 0
        event-loop-threads = ${?CONNECTIVITY_MQTT_EVENT_LOOP_THREADS}

        ## The following configuration settings are used as defaults for "specificConfic" settings of each created MQTT
        ##  connection - if not specified otherwise, those defaults are applied.

        # Indicates whether subscriber CONN messages should set clean-session or clean-start flag to true.
        clean-session = false
        clean-session = ${?CONNECTIVITY_MQTT_CLEAN_SESSION}

        # Indicates whether the client should reconnect to enforce a redelivery for a failed acknowledgement.
        reconnect-for-redelivery = false
        reconnect-for-redelivery = ${?CONNECTIVITY_MQTT_RECONNECT_FOR_REDELIVERY}

        # The amount of time that a reconnect will be delayed after a failed acknowledgement
        reconnect-for-redelivery-delay = 10s
        reconnect-for-redelivery-delay = ${?CONNECTIVITY_MQTT_RECONNECT_FOR_REDELIVERY_DELAY}

        # Indicates whether a separate client should be used for publishing. This could be useful when
        # reconnect-for-redelivery is set to true to avoid that the publisher has downtimes.
        separate-publisher-client = false
        separate-publisher-client = ${?CONNECTIVITY_MQTT_SEPARATE_PUBLISHER_CLIENT}

        reconnect {
          # The minimum reconnect timeout for when the MQTT broker closed the MQTT connection.
          min-timeout-for-mqtt-broker-initiated-disconnect = 1s
          min-timeout-for-mqtt-broker-initiated-disconnect = ${?CONNECTIVITY_MQTT_RECONNECT_MIN_TIMEOUT_FOR_MQTT_BROKER_INITIATED_DISCONNECT}

          backoff.timeout {
            # the minimum timeout for MQTT client reconnections - "0s" reconnects immediately
            min-timeout = 0s
            min-timeout = ${?CONNECTIVITY_MQTT_RECONNECT_BACKOFF_TIMEOUT_MIN}

            # the maximum timeout, once expontential backoff reached this max, the max is used instead
            max-timeout = 1m
            max-timeout = ${?CONNECTIVITY_MQTT_RECONNECT_BACKOFF_TIMEOUT_MAX}
          }
        }
      }

      http-push {
        # How many messages to buffer in the publisher actor before dropping them. Each takes up to 100 KB heap space.
        max-queue-size = 1000
        max-queue-size = ${?CONNECTIVITY_HTTP_PUSH_MAX_QUEUE_SIZE}

        request-timeout = 60s
        request-timeout = ${?CONNECTIVITY_HTTP_PUSH_REQUEST_TIMEOUT}

        # proxy config
        proxy {
          enabled = false
          enabled = ${?CONNECTIVITY_HTTP_PROXY_ENABLED}

          hostname = ${?CONNECTIVITY_HTTP_PROXY_HOST}
          port = ${?CONNECTIVITY_HTTP_PROXY_PORT}
          username = ${?CONNECTIVITY_HTTP_PROXY_USERNAME}
          password = ${?CONNECTIVITY_HTTP_PROXY_PASSWORD}
        }

        hmac-algorithms {
          aws4-hmac-sha256 = "org.eclipse.ditto.connectivity.service.messaging.httppush.AwsRequestSigningFactory"
          az-monitor-2016-04-01 =
            "org.eclipse.ditto.connectivity.service.messaging.httppush.AzMonitorRequestSigningFactory"
          az-sasl = "org.eclipse.ditto.connectivity.service.messaging.signing.AzSaslSigningFactory"
        }

        oauth2 {
          token-cache {
            # how many tokens to cache
            maximum-size = 1
            maximum-size = ${?CONNECTIVITY_HTTP_OAUTH2_CACHE_SIZE}
          }

          # Maximum clock skew of OAuth2 token endpoints. Access tokens are refreshed this long before expiration.
          max-clock-skew = 60s
          max-clock-skew = ${?CONNECTIVITY_HTTP_OAUTH2_MAX_CLOCK_SKEW}

          # Whether to enforce HTTPS for OAuth2 token endpoints. Should be `true` for production environments
          # in order not to transmit client secrets in plain text.
          enforce-https = true
          enforce-https = ${?CONNECTIVITY_HTTP_OAUTH2_ENFORCE_HTTPS}
        }

        omit-request-body-methods = ["GET", "DELETE"]
        omit-request-body-methods = ${?CONNECTIVITY_HTTP_OMIT_REQUEST_BODY_METHODS}
      }

      kafka {
        consumer {
          throttling {
            # even if Kafka trottling is disabled, the "Max inflight" messages will still be determined with `limit`
            # and `max-in-flight-factor` configuration values - so increase them if you need more "max inflight"
            # processed messages in Ditto
            enabled = false
            enabled = ${?KAFKA_CONSUMER_THROTTLING_ENABLED}

            # Interval at which the consumer is throttled. Must be > 0s.
            interval = 1s
            interval = ${?KAFKA_CONSUMER_THROTTLING_INTERVAL}

            # The maximum number of messages the consumer is allowed to receive within the configured
            # throttling interval e.g. 100 msgs/s. Must be > 0.
            limit = 100
            limit = ${?KAFKA_CONSUMER_THROTTLING_LIMIT}

            # Throttling detection tolerance in percent e.g. for a tolerance of 5% if consumed messages are above
            # limit - 5% per interval the source is considered throttled.
            throttling-detection-tolerance = 0.05
            throttling-detection-tolerance = ${?KAFKA_CONSUMER_THROTTLING_DETECTION_TOLERANCE}

            # How many unacknowledged messages are allowed at any time as factor of ${limit} - must be >= 1.0
            # This limit couples latency with throughput (long latency before ack -> lower throughput)
            max-in-flight-factor = 2.0
            max-in-flight-factor = ${?KAFKA_CONSUMER_THROTTLING_MAX_IN_FLIGHT_FACTOR}
          }

          restart {
            exponential-backoff {
              min = 100ms
              min = ${?KAFKA_CONSUMER_RESTART_BACK_OFF_MIN}
              max = 60s
              min = ${?KAFKA_CONSUMER_RESTART_BACK_OFF_MAX}
              random-factor = 1.0
              random-factor = ${?KAFKA_CONSUMER_RESTART_BACK_OFF_RANDOM_FACTOR}
            }
          }

          alpakka = ${akka.kafka.consumer} # resolve defaults from reference.conf
          alpakka = {
            use-dispatcher = "kafka-consumer-dispatcher"
            poll-interval = 50ms
            stop-timeout = 0s

            connection-checker {
              # Flag to turn on connection checker
              enable = true

              # Amount of attempts to be performed after a first connection failure occurs
              # Required, non-negative integer
              max-retries = 1

              # Interval for the connection check. Used as the base for exponential retry.
              check-interval = 5s

              # Check interval multiplier for backoff interval
              # Required, positive number
              backoff-factor = 1.0
            }

            # Properties defined by org.apache.kafka.clients.consumer.ConsumerConfig
            # can be defined in this configuration section.
            kafka-clients {
              enable.auto.commit = true
              fetch.max.wait.ms = 2500
              fetch.max.wait.ms = ${?KAFKA_CONSUMER_FETCH_MAX_WAIT_MS}
              metadata.max.age.ms = 180000
              metadata.max.age.ms = ${?KAFKA_CONNECTION_MAX_IDLE_MS}
              connections.max.idle.ms = 180000
              connections.max.idle.ms = ${?KAFKA_CONNECTION_MAX_IDLE_MS}

              # The timeout used to detect client failures when using Kafka's group management facility.
              # The client sends periodic heartbeats to indicate its liveness to the broker.
              # If no heartbeats are received by the broker before the expiration of this session timeout,
              # then the broker will remove this client from the group and initiate a rebalance.
              session.timeout.ms = 10000
              session.timeout.ms = ${?KAFKA_SESSION_TIMEOUT_MS}

              # The expected time between heartbeats to the consumer coordinator when using Kafka's group management facilities.
              # Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing
              # when new consumers join or leave the group.
              # The value must be set lower than `session.timeout.ms`, but typically should be set no higher
              # than 1/3 of that value:
              heartbeat.interval.ms = 3000
              heartbeat.interval.ms = ${?KAFKA_HEARTBEAT_INTERVAL_MS}
            }
          }
          metric-collecting-interval = 10s
          metric-collecting-interval = ${?KAFKA_METRIC_COLLECTING_INTERVAL}

          # Timeout before the consumer is initialized and considered "ready".
          init-timeout-seconds = 3
          init-timeout-seconds = ${?KAFKA_CONSUMER_INIT_TIMEOUT_SECONDS}
        }

        committer {
          alpakka = ${akka.kafka.committer} # resolve defaults from reference.conf
          alpakka = {
            # Maximum number of messages in a single commit batch
            max-batch = 1000
            max-batch = ${?KAFKA_COMMITTER_MAX_BATCH}

            # Maximum interval between commits
            max-interval = 10s
            max-interval = ${?KAFKA_COMMITTER_MAX_INTERVAL}
          }
        }

        producer {
          # If a message can't be published it is put in a queue. Further messages are dropped when the queue is full.
          queue-size = 1000
          queue-size = ${?KAFKA_PRODUCER_QUEUE_SIZE}

          # Messages to publish in parallel per Kafka-Publisher (one per connectivity client)
          parallelism = 10
          parallelism = ${?KAFKA_PRODUCER_PARALLELISM}

          # Minimum duration before restarting the Kafka-Publisher stream
          min-backoff = 1s
          min-backoff = ${?KAFKA_PRODUCER_MIN_BACKOFF}

          # Maximum duration before restarting the Kafka-Publisher stream
          max-backoff = 10s
          max-backoff = ${?KAFKA_PRODUCER_MAX_BACKOFF}

          # Random factor to vary the restart backoff
          random-factor = 0.2
          random-factor = ${?KAFKA_PRODUCER_RANDOM_FACTOR}

          # Maximum restarts within duration before failing the Kafka-Publisher stream
          max-restarts-count = 3
          max-restarts-count = ${?KAFKA_PRODUCER_MAX_RESTARTS_COUNT}
          max-restarts-within = 90s
          max-restarts-within = ${?KAFKA_PRODUCER_MAX_RESTARTS_WITHIN}

          # Timeout before the producer is initialized and considered "ready".
          init-timeout-seconds = 3
          init-timeout-seconds = ${?KAFKA_PRODUCER_INIT_TIMEOUT_SECONDS}

          alpakka = ${akka.kafka.producer} # resolve defaults from reference.conf
          alpakka {
            use-dispatcher = "kafka-producer-dispatcher"

            # Properties defined by org.apache.kafka.clients.producer.ProducerConfig
            # can be defined in this configuration section.
            kafka-clients {
              # Close idle connections after the number of milliseconds specified by this config.
              # When a message should be produced after a connection was closed because of this timeout, the client
              # simply opens the connection again, so for producers there is no need to increase this value:
              connections.max.idle.ms = 180000
              connections.max.idle.ms = ${?KAFKA_CONNECTION_MAX_IDLE_MS}

              metadata.max.age.ms = 180000
              metadata.max.age.ms = ${?KAFKA_CONNECTION_MAX_IDLE_MS}

              # The maximum amount of time in milliseconds to wait when reconnecting to a broker that has repeatedly failed to connect.
              # If provided, the backoff per host will increase exponentially for each consecutive connection failure, up to this maximum.
              reconnect.backoff.max.ms = 10000 # default: 1000
              # The base amount of time to wait before attempting to reconnect to a given host.
              # This avoids repeatedly connecting to a host in a tight loop.
              reconnect.backoff.ms = 500 # default: 50

              # Request acknowledgement
              acks = "1"

              # Disable automatic retry
              retries = 0

              # Max wait before 1 send attempt fails
              request.timeout.ms = 10000 # default: 40000

              # Max wait before send fails
              delivery.timeout.ms = 10000 # default: 120000

              # Max wait for downed broker before connection fails
              max.block.ms = 10000 # default: 60000
            }
          }
        }
      }

      cleanup {
        enabled = true
        enabled = ${?CLEANUP_ENABLED}

        quiet-period = 5m
        quiet-period = ${?CLEANUP_QUIET_PERIOD}

        interval = 60s
        interval = ${?CLEANUP_INTERVAL}

        timer-threshold = 150ms
        timer-threshold = ${?CLEANUP_TIMER_THRESHOLD}

        credits-per-batch = 1
        credits-per-batch = ${?CLEANUP_CREDITS_PER_BATCH}

        reads-per-query = 100
        reads-per-query = ${?CLEANUP_READS_PER_QUERY}

        writes-per-credit = 100
        writes-per-credit = ${?CLEANUP_WRITES_PER_CREDIT}

        delete-final-deleted-snapshot = false
        delete-final-deleted-snapshot = ${?CLEANUP_DELETE_FINAL_DELETED_SNAPSHOT}
      }
    }

    mapping {

      # the buffer size used for the queue in the message mapping processor actor
      buffer-size = 500
      buffer-size = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_BUFFER_SIZE}

      # parallelism to use for signal enriching a single message in the message mapping processor actor
      # when configured too low, throughput of signal enrichment will be low
      parallelism = 100
      parallelism = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_PARALLELISM}

      # maximum parallelism for mapping incoming and outgoing messages for a connection. This setting limits the
      # poolSize that users can configure on their connections.
      max-pool-size = 5
      max-pool-size = ${?CONNECTIVITY_MESSAGE_MAPPING_MAX_POOL_SIZE}

      # Whether messages with failed enrichments should be published.
      publish-failed-enrichments = false
      publish-failed-enrichments = ${?CONNECTIVITY_MESSAGE_MAPPING_PUBLISH_FAILED_ENRICHMENTS}

      javascript {
        # the maximum script size in bytes of a mapping script to run
        # prevents loading big JS dependencies into the script (e.g. jQuery which has ~250kB)
        maxScriptSizeBytes = 50000 # 50kB
        maxScriptSizeBytes = ${?CONNECTIVITY_MESSAGE_MAPPING_JS_MAX_SCRIPT_SIZE_BYTES}
        # the maximum execution time of a mapping script to run
        # prevents endless loops and too complex scripts
        maxScriptExecutionTime = 500ms
        maxScriptExecutionTime = ${?CONNECTIVITY_MESSAGE_MAPPING_JS_MAX_EXECUTION_TIME}
        # the maximum call stack depth in the mapping script
        # prevents recursions or other too complex computation
        maxScriptStackDepth = 25
        maxScriptStackDepth = ${?CONNECTIVITY_MESSAGE_MAPPING_JS_MAX_STACK_DEPTH}
        # Whether to allow using 'print', 'exit', 'quit' in JavaScript executions, only intended for debugging purposes
        allowUnsafeStandardObjects = false
        allowUnsafeStandardObjects = ${?CONNECTIVITY_MESSAGE_MAPPING_JS_ALLOW_UNSAFE_STANDARD_OBJECTS}
        # The filesystem path where to load CommonJS modules from, by default empty indicating to not load any CommonJS modules
        commonJsModulePath = ""
        commonJsModulePath = ${?CONNECTIVITY_MESSAGE_MAPPING_JS_COMMON_JS_MODULE_PATH}
      }

      mapper-limits {
        # maximum number of mappers defined in one source
        max-source-mappers = 10
        # maximum number of messages invoked by a mapper
        # defined in source
        max-mapped-inbound-messages = 10
        # maximum number of mappers defined in one source
        max-target-mappers = 10
        # maximum number of messages invoked by a mapper
        # defined in target
        max-mapped-outbound-messages = 10
      }
    }

    signal-enrichment {
      // Beware: Despite similarities with gateway signal-enrichment providers,
      // this class is different and not compatible with them.
      provider = "org.eclipse.ditto.connectivity.service.mapping.ConnectivityCachingSignalEnrichmentProvider"
      provider = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_PROVIDER}

      provider-config {
        # timeout for all facades
        ask-timeout = 10s
        ask-timeout = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_ASK_TIMEOUT}

        cache {
          # how many things to cache in total on a single cluster node
          maximum-size = 20000
          maximum-size = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_CACHE_MAXIMUM_SIZE}

          # maximum duration of inconsistency after e.g. a policy update
          expire-after-create = 2m
          expire-after-create = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_CACHE_EXPIRE_AFTER_CREATE}
        }
      }
      # Which caching signal enrichment facade is used by the signal-enrichment.provider
      caching-signal-enrichment-facade.provider = ${?CONNECTIVITY_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
    }

    persistence-ping {
      # journal tag to query to find our which connectionPersistenceActors to ping (reconnect)
      journal-tag = ""
      journal-tag = ${?RECONNECT_JOURNAL_TAG}
      # initial delay for pinging / reconnecting the connections after the PersistencePingActor has been started.
      initial-delay = 0s
      initial-delay = ${?RECONNECT_INITIAL_DELAY}
      # interval for trying to reconnect all started connections.
      interval = 10m
      interval = ${?RECONNECT_INTERVAL}
      # how many events to read in one query
      read-journal-batch-size = 500
      read-journal-batch-size = ${?RECONNECT_READ_JOURNAL_BATCH_SIZE}

      # one of:
      #  TAGS  Elements will be ordered by their tags.
      #  ID    Elements will be ordered by their document ID.
      streaming-order = TAGS
      streaming-order = ${?PERSISTENCE_PING_STREAMING_ORDER}

      # used to throttle pinging of connections, so that not all connections are recovered at the same time
      rate {
        frequency = 1s
        frequency = ${?RECONNECT_RATE_FREQUENCY}
        entities = 1
        entities = ${?RECONNECT_RATE_ENTITIES}
      }
    }

    connection-ids-retrieval {
      # how many events to read in one query
      read-journal-batch-size = 500
      read-journal-batch-size = ${?CONNECTION_IDS_RETRIEVAL_READ_JOURNAL_BATCH_SIZE}
      # how many snapshots to read in one query
      read-snapshot-batch-size = 50
      read-snapshot-batch-size = ${?CONNECTION_IDS_RETRIEVAL_READ_SNAPSHOT_BATCH_SIZE}
    }

    client {
      # Initial timeout after the init process is triggered.
      init-timeout = 5s
      init-timeout = ${?CONNECTIVITY_CLIENT_INIT_TIMEOUT}
      # Initial timeout when connecting to a remote system. If the connection could not be established after this time,
      # the service will try to reconnect. If a failure happened during connecting, then the service will wait
      # for at least this time until it will try to reconnect. The max timeout is defined in connecting-max-timeout.
      # this timeout needs to be smaller than ditto.connectivity.connection.client-actor-ask-timeout in connectivity.conf
      connecting-min-timeout = 50s
      connecting-min-timeout = ${?CONNECTIVITY_CLIENT_CONNECTING_MIN_TIMEOUT}
      # Max timeout (until reconnecting) when connecting to a remote system.
      # Should be greater than connecting-min-timeout.
      connecting-max-timeout = 60m
      connecting-max-timeout = ${?CONNECTIVITY_CLIENT_CONNECTING_MAX_TIMEOUT}
      # How many times we will try to reconnect when connecting to a remote system.
      # max time ~= connecting-max-tries * connecting-max-timeout = 50 * 60m = 50h
      connecting-max-tries = 50
      connecting-max-tries = ${?CONNECTIVITY_CLIENT_CONNECTING_MAX_TRIES}
      # Max timeout when actually disconnecting until we're assuming disconnecting failed. The timeout will start
      # after a disconnect announcment was sent (see disconnect-announcement-timeout) and disconnecting is actually
      # triggered. Thus, the total amount of timeout that might elapse between closing a connection and running into
      # the timeout, is: disconnect-announcement-timeout + disconnecting-max-timeout.
      disconnecting-max-timeout = 5s
      disconnecting-max-timeout = ${?CONNECTIVITY_CLIENT_DISCONNECTING_MAX_TIMEOUT}
      # Time that will be waited between sending a disconnect announcement and actually disconnecting.
      # When closing a connection, a disconnect announcement (special topic for connection targets) will be sent to
      # the client. The connection will wait for disconnect-announcement-timeout before actually disconnecting
      # the client.
      disconnect-announcement-timeout = 3s
      disconnect-announcement-timeout = ${?CONNECTIVITY_CLIENT_DISCONNECT_ANNOUNCEMENT_TIMEOUT}
      # Max timeout until waiting for search commands.
      subscription-manager-timeout = 60s
      subscription-manager-timeout = ${?CONNECTIVITY_CLIENT_SUBSCRIPTION_MANAGER_TIMEOUT}
      # how long the service will wait for a successful connection when testing a new connection. If no response is
      # received after this duration, the test will be assumed a failure.
      testing-timeout = 10s
      testing-timeout = ${?CONNECTIVITY_CLIENT_TESTING_TIMEOUT}
      # Min backoff after connection failure.
      min-backoff = 5s
      min-backoff = ${?CONNECTIVITY_CLIENT_MIN_BACKOFF}
      # Max backoff after connection failure.
      max-backoff = 60m
      max-backoff = ${?CONNECTIVITY_CLIENT_MAX_BACKOFF}
      # How often to refresh the cache of other client actors if client count > 1
      client-actor-refs-notification-delay = 5m
    }

    monitoring {
      logger {
        successCapacity = 10
        successCapacity = ${?CONNECTIVITY_LOGGER_SUCCESS_CAPACITY}
        failureCapacity = 10
        failureCapacity = ${?CONNECTIVITY_LOGGER_FAILURE_CAPACITY}
        maxLogSizeBytes = 250000
        maxLogSizeBytes = ${?CONNECTIVITY_LOGGER_MAX_LOG_SIZE_BYTES}
        logDuration = 1h
        logDuration = ${?CONNECTIVITY_LOGGER_LOG_DURATION}
        loggingActiveCheckInterval = 5m
        loggingActiveCheckInterval = ${?CONNECTIVITY_LOGGER_ACTIVE_CHECK_INTERVAL}

        # configuration for FluentPublishingConnectionLogger used to publish connection logs to a fluentd/fluentbit endpoint
        publisher {
          enabled = false
          enabled = ${?CONNECTIVITY_LOGGER_PUBLISHER_ENABLED}

          # the log levels to use for logging in the publisher as comma separated list
          # supported: "failure" "success"
          logLevels = "success,failure"
          logLevels = ${?CONNECTIVITY_LOGGER_PUBLISHER_LOG_LEVELS}

          # whether to log headers and payload for publisher logs:
          logHeadersAndPayload = false
          logHeadersAndPayload = ${?CONNECTIVITY_LOGGER_PUBLISHER_LOG_HEADERS_AND_PAYLOAD}

          # a specific log-tag to use for the published logs. If null, a default log-tag will be used: "connection:<connection-id>"
          log-tag = null
          log-tag = ${?CONNECTIVITY_LOGGER_PUBLISHER_LOG_TAG}

          # add key/value pairs to this section in order to have them added to published logs
          additional-log-context {
            # by default empty
          }

          # fluency is the library used to publish connection logs to a fluentd/fluentbit endpoint
          fluency {
            ## NetworkSender.Config
            host = "localhost"
            host = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_HOST}

            port = 24224
            port = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_PORT}

            sslEnabled = false            # library default: false
            sslEnabled = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_SSL_ENABLED}

            # Socket connection timeout:
            connectionTimeout = 5s        # library default: 5s, min: 10ms
            connectionTimeout = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_CONNECTION_TIMEOUT}

            # Socket read timeout:
            readTimeout = 5s              # library default: 5s, min: 10ms
            readTimeout = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_READ_TIMEOUT}

            ## Flusher.Config

            # how often to check if a flush was requested from "outside", e.g. by closing the flusher
            # and send logs to the endpoint if that was the case:
            flushAttemptInterval = 600ms  # library default: 600ms, min: 20ms, max: 2000ms

            # when closing flusher, wait that long until forcefully stopping flusher:
            waitUntilBufferFlushed = 60s  # library default: 60s, min: 1s

            # when closing flusher, wait that long for resoruces until forcefully stopping:
            waitUntilTerminated = 60s     # library default: 60s, min: 1s

            ## Buffer.Config

            # the maximum possible log entries total size to buffer in memory:
            maxBufferSize = 67108864            # 64 * 1024 * 1024 - library default: 512 * 1024 * 1024 = 512MiB

            # the initial chunk size, that much memory is allocated at least per instance:
            bufferChunkInitialSize = 1048576    # library default: 1024 * 1024 = 1MiB

            # after the buffer reached this size, perform a flush, sending logs to the endpoint:
            bufferChunkRetentionSize = 4194304  # library default: 4 * 1024 * 1024 = 4MiB

            # after the oldest buffered entry is this old, perform a flush, sending logs to the endpoint:
            bufferChunkRetentionTime = 1s       # library default: 1s

            # Use JVM heap memory for buffer pool, if `false`, memory is allocated via `ByteBuffer.allocateDirect`
            jvmHeapBufferMode = false           # library default: false

            # for file backup mode:
            # "In this mode, Fluency takes backup of unsent memory buffers as files when closing and then resends them when restarting"
            fileBackupDir = null                # library default: null

            ## ExponentialBackOffRetryStrategy.Config

            senderMaxRetryCount = 7   # min: 0 (meaning retries off)

            ## RetryStrategy.Config

            # retry base interval:
            senderBaseRetryInterval = 400ms  # min: 10ms library default: 400ms

            # retry max interval:
            senderMaxRetryInterval = 30s     # min: 10ms library default: 30s

            ## FluentdIngester.Config

            # whether ACK response mode is enabled, meaning that it is waited for the ACK of the fluentd/fluentbit
            # instance where the logs are published to. Enables "at-least-once" semantics:
            ackResponseMode = false         # library default: false

            # The duration of how long to wait after closing the Fluency buffer.
            # If this is Zero or a negative duration, no waiting for the buffer will be performed:
            waitUntilAllBufferFlushedDurationOnClose = 5s
            waitUntilAllBufferFlushedDurationOnClose = ${?CONNECTIVITY_LOGGER_PUBLISHER_FLUENCY_WAIT_UNTIL_ALL_BUFFER_FLUSHED_DURATION_ON_CLOSE}
          }
        }
      }
      counter {}
    }

    default-config-provider = true
    default-config-provider = ${?CONNECTIVITY_DEFAULT_CONFIG_PROVIDER}

    tunnel {
      # 0 workers mean the sshd client defaults which are: `Runtime.getRuntime().availableProcessors() + 1`
      workers = 0
      heartbeat-interval = 30s
      # by default, keep the tunnel open forever:
      idle-timeout = 0s
      socket-keepalive = true
    }

  }
}

akka {
  http {
    client {
      user-agent-header = eclipse-ditto/${ditto.version}

      # Setting this to less than Inf may generate occasional failures for not very active HTTP-PUSH connections.
      idle-timeout = Inf
      idle-timeout = ${?AKKA_HTTP_CLIENT_IDLE_TIMEOUT}
    }
    # The connection-pools used for http-push connections
    host-connection-pool {
      # This timeout configures a maximum amount of time, while the connection can be kept open.
      # Helps the process of rebalancing between service instances when reaching the target through a load-balancer
      max-connection-lifetime = Inf
      max-connection-lifetime = ${?AKKA_HTTP_HOSTPOOL_MAX_CONNECTION_LIFETIME}
    }
  }
  cluster {
    sharding {
      role = ${ditto.service-name}

      # When this is set to 'on' the active entity actors will automatically be restarted
      # upon Shard restart. i.e. if the Shard is started on a different ShardRegion
      # due to rebalance or crash.
      remember-entities = on
      # increase default timeout because otherwise shards are beeing stopped and restarted if the ShardCoordinator
      # could not load initial entity ids from the "remember entities store" within this timeout. This is required
      # since remember-entities is set to "on".
      waiting-for-state-timeout = 10s
      passivation {
        # don't passivate shards by default as Ditto AbstractShardedPersistenceActor decides that on its own -
        # default is "default-idle-strategy" with "idle-entity.timeout = 120s":
        strategy = "off"
      }

      # Timeout of waiting for update the distributed state (update will be retried if the timeout happened)
      # Also used as timeout for writes of remember entities when that is enabled
      updating-state-timeout = 15s

      entity-recovery-strategy = "constant"
      entity-recovery-constant-rate-strategy {
        # Sets the frequency at which a batch of entity actors is started.
        frequency = 100 ms
        # Sets the number of entity actors to be restart at a particular interval
        number-of-entities = 10
      }
    }

    roles = [
      "connectivity"
    ]
  }
  coordinated-shutdown {
    phases {
      before-service-unbind {
        # default timeout is 5s for the phase - give a longer timeout in order
        # to be able to send disconnect announcements of connections:
        timeout = 10s
        timeout = ${?AKKA_COORDINATED_SHUTDOWN_PHASES_BEFORE_SERVICE_UNBIND_TIMEOUT}
      }
    }
  }
  persistence {
    journal.auto-start-journals = [
      "akka-contrib-mongodb-persistence-connection-journal"
    ]
    snapshot-store.auto-start-snapshot-stores = [
      "akka-contrib-mongodb-persistence-connection-snapshots"
    ]
  }
}

include "ditto-protocol-subscriber.conf"

akka-contrib-mongodb-persistence-connection-journal {
  class = "akka.contrib.persistence.mongodb.MongoJournal"
  plugin-dispatcher = "connection-persistence-dispatcher"

  circuit-breaker {
    max-failures = 5 # if an exception during persisting an event/snapshot occurs this often -- a successful write resets the counter
    max-failures = ${?BREAKER_MAXTRIES}
    call-timeout = 5s # MongoDB Timeouts causing the circuitBreaker to open
    call-timeout = ${?BREAKER_TIMEOUT}
    reset-timeout = 3s # after this time in "Open" state, the cicuitBreaker is "Half-opened" again
    reset-timeout = ${?BREAKER_RESET}
  }

  overrides {
    journal-collection = "connection_journal"
    journal-index = "connection_journal_index"

    realtime-collection = "connection_realtime"
    metadata-collection = "connection_metadata"
  }

  event-adapters {
    mongodbobject = "org.eclipse.ditto.connectivity.service.messaging.persistence.ConnectivityMongoEventAdapter"
  }

  event-adapter-bindings {
    "org.eclipse.ditto.base.model.signals.events.Event" = mongodbobject
    "org.bson.BsonValue" = mongodbobject
  }
}

akka-contrib-mongodb-persistence-connection-snapshots {
  class = "akka.contrib.persistence.mongodb.MongoSnapshots"
  plugin-dispatcher = "connection-persistence-dispatcher"
  overrides {
    snaps-collection = "connection_snaps"
    snaps-index = "connection_snaps_index"
  }
}

connection-persistence-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 4
    parallelism-factor = 3.0
    parallelism-max = 32
    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
  }
  # Throughput defines the maximum number of messages to be
  # processed per actor before the thread jumps to the next actor.
  # Set to 1 for as fair as possible.
  throughput = 2
}

rabbit-stats-bounded-mailbox {
  mailbox-type = "akka.dispatch.BoundedMailbox"
  mailbox-capacity = 10
  mailbox-push-timeout-time = 0s
}

message-mapping-processor-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  fork-join-executor {
    # Min number of threads to cap factor-based parallelism number to
    parallelism-min = 4
    # Parallelism (threads) ... ceil(available processors * factor)
    parallelism-factor = 3.0
    # Max number of threads to cap factor-based parallelism number to
    parallelism-max = 64
  }
  throughput = 5
}

jms-connection-handling-dispatcher {
  # one thread per actor because the actor blocks.
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

signal-enrichment-cache-dispatcher {
  type = "Dispatcher"
  executor = "thread-pool-executor"
  thread-pool-executor {
    keep-alive-time = 60s
    fixed-pool-size = off
    max-pool-size-max = 256
    max-pool-size-max = ${?CACHE_DISPATCHER_POOL_SIZE_MAX}
    max-pool-size-max = ${?SIGNAL_ENRICHMENT_CACHE_DISPATCHER_POOL_SIZE_MAX}
  }
}

http-push-connection-dispatcher {
  type = "Dispatcher"
  executor = "thread-pool-executor"
  thread-pool-executor {
    keep-alive-time = 60s
    fixed-pool-size = off
    max-pool-size-max = 256
    max-pool-size-max = ${?HTTP_PUSH_CONNECTION_DISPATCHER_POOL_SIZE_MAX}
  }
}

kafka-consumer-dispatcher {
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

kafka-producer-dispatcher {
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

include "connectivity-extension"