test.conf

akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "DEBUG"
  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"

  # for log messages during the actor system is starting up and shutting down:
  stdout-loglevel = "WARNING"

  log-config-on-start = off

  http {
    client.idle-timeout = Inf
  }
}

// dead lettters log disabled because actors are being killed all the time
akka.log-dead-letters = 0
akka.cluster.jmx.multi-mbeans-in-same-jvm = on
akka.cluster.roles = ["thing-event-aware", "live-signal-aware", "acks-aware", "policy-announcement-aware"]

ditto {
  mapping-strategy.implementation = "org.eclipse.ditto.connectivity.api.ConnectivityMappingStrategies"

  ddata {
    subscription-write-consistency = "local"
    subscription-delay = "0s"
  }

  pubsub {
    restart-delay = 1s
    update-interval = 100ms // increase this value to simulate slow systems
    seed = "dummy-seed"
  }

  mongodb {
    uri = "mongodb://localhost:27017/connectivity"
  }
  connectivity {
    default-config-provider = true

    user-indicated-errors = [
      {exceptionName: "org.apache.qpid.jms.provider.exceptions.ProviderSecurityException", messagePattern: ".*"}
    ]
    connection {
      // allow localhost in unit tests
      blocked-hostnames = ""

      # Number of sources per connection
      max-source-number = 4
      # Number of targets per connection
      max-target-number = 4

      client-actor-restarts-before-escalation = 3

      supervisor {
        exponential-backoff {
          min = 1s
          max = 5s
          random-factor = 1.0
        }
      }

      snapshot {
        threshold = 5
      }

      activity-check {
        # the interval of how long to keep an "inactive" Connection in memory:
        inactive-interval = 0 # keep active indefinitely

        # the interval of how long to keep a deleted Connection in memory:
        deleted-interval = 1s
      }

      flush-pending-responses-timeout = 0s

      kafka {
        consumer {
          metric-collecting-interval = 5s
          init-timeout-seconds = 3
          alpakka = ${akka.kafka.consumer}
          alpakka = {
            default.key.serde = "org.apache.kafka.common.serialization.Serdes$StringSerde" # default:	org.apache.kafka.common.serialization.Serdes$ByteArraySerde
            default.value.serde = "org.apache.kafka.common.serialization.Serdes$StringSerde" # default:	org.apache.kafka.common.serialization.Serdes$ByteArraySerde
            kafka-clients {
              enable.auto.commit = true
              retries = 0
              request {
                timeout {
                  ms = 10000
                }
              }
            }
          }
        }

        producer {
          queue-size = 39
          parallelism = 3
          min-backoff = 3s
          max-backoff = 30s
          random-factor = 0.2
          init-timeout-seconds = 3

          alpakka = ${akka.kafka.producer}
          alpakka {
            kafka-clients {
              connections.max.idle.ms = 543210
              reconnect.backoff {
                ms = 500
                max {
                  ms = 10000
                }
              }
            }
          }
        }

      }

      amqp10 {
        consumer {
          throttling {
            interval = 100ms
            limit = 1
          }
        }
        publisher {
          max-queue-size = 2
          parallelism = 2
        }

        producer-cache-size = 10
        global-send-timeout = 120s

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

        include "backoff-test"
      }

      amqp091 {
        publisher {
          pending-ack-ttl = 1d
        }
      }

      http-push {
        max-queue-size = 100
        proxy {
          enabled = false
          hostname = ""
          port = 0
          username = ""
          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"
        }
      }

      mqtt {
        reconnect-for-redelivery = true
        separate-publisher-client = true

        reconnect {
          backoff {

          }
        }
      }

    }

    mapping {
      buffer-size = 10
      parallelism = 1
      javascript {
        maxScriptSizeBytes = 50000 # 50kB
        maxScriptExecutionTime = 30000ms
        maxScriptStackDepth = 25
        commonJsModulePath = "./target/test-classes/unpacked-test-webjars"
      }
      mapper-limits {
        max-source-mappers = 9
        max-mapped-inbound-messages = 8
        max-target-mappers = 7
        max-mapped-outbound-messages = 6
      }
    }

    signal-enrichment {
      provider = "org.eclipse.ditto.connectivity.service.mapping.ConnectivityByRoundTripSignalEnrichmentProvider"
      provider-config {
        ask-timeout = 10s
        // configure cache here in case signal enrichment provider gets swapped to the caching implementation
        cache {
          maximum-size = 1000
          expire-after-create = 2m
        }
      }
    }

    persistence-ping {
      # 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}

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

    client {
      connecting-min-timeout = 5s
      connecting-max-timeout = 5s
      connecting-max-tries = 10
      disconnecting-max-timeout = 3s
      disconnect-announcement-timeout = 1s
      testing-timeout = 5s
      min-backoff = 100ms
      max-backoff = 400ms
    }

    monitoring {

      logger {
        successCapacity = 3
        successCapacity = ${?CONNECTIVITY_LOGGER_SUCCESS_CAPACITY}
        failureCapacity = 10
        failureCapacity = ${?CONNECTIVITY_LOGGER_FAILURE_CAPACITY}
        maxLogSizeBytes = 1000
        maxLogSizeBytes = ${?CONNECTIVITY_LOGGER_MAX_LOG_SIZE_BYTES}
        logDuration = 10m
        logDuration = ${?CONNECTIVITY_LOGGER_LOG_DURATION}
        loggingActiveCheckInterval = 1m
      }

      counter {}
    }

  }
}

akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "DEBUG"
  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"

  # for log messages during the actor system is starting up and shutting down:
  stdout-loglevel = "WARNING"

  log-config-on-start = off

  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
    enable-additional-serialization-bindings = on

    # this is only intended for testing.
    serialize-messages = off
    serialize-creators = off

    debug {
      lifecycle = on
    }

    default-dispatcher {
      fork-join-executor {
        parallelism-min = 4
        parallelism-factor = 3.0
        parallelism-max = 32
        parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
      }
    }

    serializers {
      bson = "org.eclipse.ditto.internal.utils.test.mongo.BsonDocumentSerializer"
      json = "org.eclipse.ditto.internal.utils.cluster.JsonJsonifiableSerializer"
      cbor = "org.eclipse.ditto.internal.utils.cluster.CborJsonifiableSerializer"
      cbor-json-value = "org.eclipse.ditto.internal.utils.cluster.CborJsonValueSerializer"
      jackson-cbor = "akka.serialization.jackson.JacksonCborSerializer"
    }

    serialization-bindings {
      "org.bson.BsonDocument" = bson
      # Serialize Jsonifiable events with custom JSON serializer:
      "org.eclipse.ditto.base.model.json.Jsonifiable" = cbor
      "org.eclipse.ditto.base.model.exceptions.DittoRuntimeException" = cbor
      "org.eclipse.ditto.base.api.devops.signals.commands.DevOpsCommandResponse" = json # to ensure readability
      "org.eclipse.ditto.json.JsonValue" = cbor-json-value
      "org.eclipse.ditto.internal.utils.cluster.AkkaJacksonCborSerializable" = jackson-cbor
    }
  }

  extensions = [
    "akka.cluster.pubsub.DistributedPubSub"
  ]

  remote {
    log-remote-lifecycle-events = on
    artery {
      enabled = on
      transport = tcp
      canonical {
        # InetAddress.getLocalHost.getHostAddress is used if empty
        hostname = "127.0.0.1"
        hostname = ${?REMOTE_HOSTNAME}
        port = 0
        port = ${?REMOTE_PORT}
      }
      bind {
        hostname = ${?BIND_HOSTNAME}
        port = ${?BIND_REMOTE_PORT}
      }
    }
  }

  cluster {
    # Disable legacy metrics in akka-cluster.
    metrics.enabled = off

    # enable weakly up feature to allow members to join even if some members are unreachable
    allow-weakly-up-members = on

    # required for akka-management-cluster-bootstrap (to be more robust):
    shutdown-after-unsuccessful-join-seed-nodes = 40s

    sharding {
      state-store-mode = ddata
      use-dispatcher = "sharding-dispatcher"

      role = "connectivity"
    }

    roles = [
      "connectivity",
      "thing-event-aware",
      "live-signal-aware",
      "acks-aware",
      "policy-announcement-aware"
    ]

    distributed-data {
      # set gossip interval and notify-subscribers-interval to low values so that pubsub delay is minimal
      gossip-interval = 50ms
      notify-subscribers-interval = 50ms
    }
  }

  http {
    server {
      server-header = "" # default: akka-http/${akka.http.version}
      max-connections = 4096 # default: 1024
      backlog = 100 # default: 100

      parsing {
        max-uri-length = 8k # default: 2k
        max-content-length = 10m # default: 8m
        illegal-header-warnings = off # default: on
        error-logging-verbosity = simple # default: full
        uri-parsing-mode = relaxed # default: strict
      }
    }

    host-connection-pool {
      # The maximum number of open requests accepted into the pool across all
      # materializations of any of its client flows.
      # Protects against (accidentally) overloading a single pool with too many client flow materializations.
      # Note that with N concurrent materializations the max number of open request in the pool
      # will never exceed N * max-connections * pipelining-limit.
      # Must be a power of 2 and > 0!
      max-open-requests = 1024 # default: 32

      # The time after which an idle connection pool (without pending requests)
      # will automatically terminate itself. Set to `infinite` to completely disable idle timeouts.
      idle-timeout = 60s # default: 30s
    }
  }

  test {
    # factor by which to scale timeouts during tests, e.g. to account for shared
    # build system load
    timefactor = 1.0

    # duration of EventFilter.intercept waits after the block is finished until
    # all required messages are received
    filter-leeway = 3s

    # duration to wait in expectMsg and friends outside of within() block
    # by default
    single-expect-default = 3s

    # The timeout that is added as an implicit by DefaultTimeout trait
    default-timeout = 5s

    calling-thread-dispatcher {
      type = akka.testkit.CallingThreadDispatcherConfigurator
    }
  }
}

akka.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.persistence.inmemory.journal.InMemoryAsyncWriteJournal"
  plugin-dispatcher = "connection-persistence-dispatcher"

  ask-timeout = 10s

  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.persistence.inmemory.snapshot.InMemorySnapshotStore"
  plugin-dispatcher = "connection-persistence-dispatcher"

  ask-timeout = 10s
}

akka-contrib-mongodb-persistence-reconnect-journal {
  class = "akka.persistence.inmemory.journal.InMemoryAsyncWriteJournal"
  plugin-dispatcher = "reconnect-persistence-dispatcher"

  ask-timeout = 10s

  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-reconnect-snapshots {
  class = "akka.persistence.inmemory.snapshot.InMemorySnapshotStore"
  plugin-dispatcher = "reconnect-persistence-dispatcher"

  ask-timeout = 10s
}

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 = 5
}

reconnect-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 = 5
}

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 = 4.0
    # Max number of threads to cap factor-based parallelism number to
    parallelism-max = 64
  }
  throughput = 5
}

jms-connection-handling-dispatcher {
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

signal-enrichment-cache-dispatcher {
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

http-push-connection-dispatcher {
  type = PinnedDispatcher
  executor = "thread-pool-executor"
}

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

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