connectivity.conf

ditto {
  mongodb {
    options {
      ssl = false
      ssl = ${?MONGO_DB_SSL_ENABLED}
      w = 1
    }
  }

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

  connectivity {
    connection {
      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 between subscribing to pub/sub topics and sending response
      flush-pending-responses-timeout = 5s
      flush-pending-responses-timeout = ${?CONNECTIVITY_FLUSH_PENDING_RESPONSES_TIMEOUT}

      # 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}

      amqp10 {
        consumer {
          throttling {
            # Interval at which the consumer is throttled. Disable throttling with a value of zero.
            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. Disable throttling with a value of zero.
            limit = 100
            limit = ${?AMQP10_CONSUMER_THROTTLING_LIMIT}
          }
        }

        // 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}
      }

      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}

        # whether Ditto should use the legacy mode MQTT connection (with alpakka-mqtt using Paho)
        legacy-mode = false
        legacy-mode = ${?CONNECTIVITY_MQTT_LEGACY_MODE}
      }

      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 = 100
        max-queue-size = ${?CONNECTIVITY_HTTP_PUSH_MAX_QUEUE_SIZE}

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

        # 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}
        }
      }

      kafka.producer.internal { # internal configuration as needed by Kafka client library
        # Tuning parameter of how many sends that can run in parallel.
        parallelism = 100

        # Duration to wait for `KafkaConsumer.close` to finish.
        close-timeout = 60s

        # Fully qualified config path which holds the dispatcher configuration
        # to be used by the producer stages. Some blocking may occur.
        # When this value is empty, the dispatcher configured for the stream
        # will be used.
        use-dispatcher = "akka.kafka.default-dispatcher"

        # The time interval to commit a transaction when using the `Transactional.sink` or `Transactional.flow`
        eos-commit-interval = 100ms

        # 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 = 540000 # default: 540000 (9min)

          # 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
        }
      }
    }

    mapping {

      factory = "org.eclipse.ditto.services.connectivity.mapping.MessageMappers"

      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
        # the maximum execution time of a mapping script to run
        # prevents endless loops and too complex scripts
        maxScriptExecutionTime = 500ms
        # the maximum call stack depth in the mapping script
        # prevents recursions or other too complex computation
        maxScriptStackDepth = 10
      }
    }

    reconnect {
      # initial delay for reconnecting the connections after the ReconnectActor 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}

      # used to throttle recovery 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}
      }
    }

    client {
      # Init timeout for client actors (if no connect msg is received the parent actor is asked whether to connect)
      # Running into init-timeout is the only way for client actors on remote nodes to start a connection.
      # Set to be as short as possible without timing out the local client actor, which is used for acknowledgement.
      init-timeout = 3s
      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.
      connecting-min-timeout = 60s
      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}
      # 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
      # 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}
    }

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

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

akka {
  cluster {
    sharding {
      role = "connectivity"

      # 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
    }

    roles = [
      "connectivity",
      "thing-event-aware",
      "live-signal-aware"
    ]
  }

  persistence {
    journal.auto-start-journals = [
      "akka-contrib-mongodb-persistence-connection-journal"
    ]
    snapshot-store.auto-start-snapshot-stores = [
      "akka-contrib-mongodb-persistence-connection-snapshots"
    ]
  }
}

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

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

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

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

  event-adapter-bindings {
    "org.eclipse.ditto.signals.events.base.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"
}

include "connectivity-extension"