ditto-akka-config.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 = "INFO"

  log-config-on-start = off

  io.dns {

    resolver = "inet-address"

    inet-address {
      # To set the time to cache name resolutions
      # Possible values:
      # default: sun.net.InetAddressCachePolicy.get() and getNegative()
      # forever: cache forever
      # never: no caching
      # n [time unit]: positive timeout with unit, for example "30 s"
      positive-ttl = never
      negative-ttl = never
    }
  }

  discovery {
    # pick the discovery method you'd like to use:
    method = akka-dns
    method = ${?DISCOVERY_METHOD}

    kubernetes-api {
      pod-label-selector = "actorSystemName=%s"
    }

    # DNS based service discovery in docker swarm (without DNS ttl caching)
    docker-swarm-dns {
      class = org.eclipse.ditto.base.service.DockerSwarmServiceDiscovery
    }
  }

  coordinated-shutdown.exit-jvm = on

  management {
    http {
      bind-hostname = "0.0.0.0"

      # enable POST, PUT and DELETE operations on Akka Mangement API
      # e.g. used to initiate full cluster shutdown via API
      route-providers-read-only = false
    }

    health-checks.readiness-checks {
      # when this is empty, the cluster-membership check is disabled for readiness:
      # cluster-membership = ""
    }

    cluster.bootstrap {

      new-cluster-enabled = on
      new-cluster-enabled = ${?CLUSTER_NEW_CLUSTER_ENABLED}

      contact-point-discovery {
        service-name = "ditto-cluster"
        service-name = ${?CLUSTER_BS_SERVICE_NAME}
        service-namespace = ${?CLUSTER_BS_SERVICE_NAMESPACE}
        effective-name = ${?CLUSTER_BS_EFFECTIVE_NAME}

        discovery-method = akka-dns
        discovery-method = ${?DISCOVERY_METHOD}
        discovery-method = ${?BOOTSTRAP_DISCOVERY_METHOD}

        required-contact-point-nr = ${?CLUSTER_BS_REQUIRED_CONTACTS}
      }
    }
  }

  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
    }

    serializers {
      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"
    }

    # Ditto custom settings:
    serializers-json {
      # The number of bytes per direct buffer in the pool used to read or write messages during JSON serialization
      direct-buffer-size = ${akka.remote.artery.advanced.maximum-frame-size}

      # The maximal number of direct buffers kept in the direct buffer pool for reuse
      direct-buffer-pool-limit = 128
    }

    serialization-bindings {
      #"java.io.Serializable" = none # must not be set in order to get akka.cluster.sharding.ShardRegion$GetShardRegionStats$ serialized
      # 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.json.JsonValue" = cbor-json-value
      "org.eclipse.ditto.internal.utils.cluster.AkkaJacksonCborSerializable" = jackson-cbor
    }

    default-dispatcher {
      executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedForkJoinExecutorServiceConfigurator"
      fork-join-executor {
        parallelism-min = 4
        parallelism-factor = 3.0
        parallelism-max = 32
        parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
      }
    }

    default-mailbox {
        mailbox-type = "org.eclipse.ditto.internal.utils.akka.mailbox.MonitoredUnboundedMailboxType"
        mailbox-type = ${?DEFAULT_MAILBOX}

        monitored-unbounded-mailbox {
          # The number of messages in a mailbox, until its size will be logged
          threshold-for-logging = 100
          threshold-for-logging = ${?MONITOR_MAILBOX_SIZE_LOGGING_THRESHOLD}
          # The minimum interval between mailbox-size log statements in nanoseconds. 1000000000ns = 1s
          logging-interval = 1000000000
          logging-interval = ${?MONITOR_MAILBOX_SIZE_LOGGING_INTERVAL}
          # A regular expression to define for which actors the mailbox size shall be tracked.
          # The regex matches agains the actors path without address, e.g.:
          # /user/gatewayRoot/fooForwarder with regex filters: /user.*
          # For the not tracked actors, the default (UnboundedMailbox.MessageQueue) will be used.
          include-actors-regex = "(\/user.*)"
          include-actors-regex = ${?MONITOR_MAILBOX_SIZE_INCLUDE_ACTORS_REGEX}
          # To explicitly exclude actors, e.g. when 'include-actor-regex' is to coarse, exclude regex can be defined.
          exclude-actors-regex = ""
          exclude-actors-regex = ${?MONITOR_MAILBOX_SIZE_EXCLUDE_ACTORS_REGEX}
        }
    }

  }

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

  remote {
    log-remote-lifecycle-events = on

    artery {
      enabled = on
      enabled = ${?ARTERY_ENABLED}
      # useful default for Ditto: "tcp" - as requires less memory, CPU, etc. than "aeron-udp"
      # (which is also more complicated to configure correctly):
      transport = tcp
      transport = ${?ARTERY_TRANSPORT}
      canonical {
        #   "<getHostAddress>"   InetAddress.getLocalHost.getHostAddress
        #   "<getHostName>"      InetAddress.getLocalHost.getHostName
        hostname = "<getHostAddress>"  # external (logical) hostname
        hostname = ${?REMOTE_HOSTNAME}
        port = 2551                   # external (logical) port
        port = ${?REMOTE_PORT}
      }

      bind {
        hostname = ""   # internal (bind) hostname -> "" means use the same as the canonical one
        hostname = ${?BIND_HOSTNAME}
        port = ""       # internal (bind) port
        port = ${?BIND_REMOTE_PORT}
      }

      advanced {
        # Maximum serialized message size, including header data. # default: 256 KiB
        maximum-frame-size = 256 KiB
        maximum-frame-size = ${?REMOTE_MAX_FRAMESIZE}
        # Direct byte buffers are reused in a pool with this maximum size.
        buffer-pool-size = 128
        # Maximum serialized message size for the large messages, including header data. # default: 2 MiB
        maximum-large-frame-size = 256 KiB
        maximum-large-frame-size = ${?REMOTE_MAX_FRAMESIZE}
        # Direct byte buffers for the large messages are reused in a pool with this maximum size.
        large-buffer-pool-size = 32

        # Total number of inbound lanes, shared among all inbound associations. A value
        # greater than 1 means that deserialization can be performed in parallel for
        # different destination actors.
        inbound-lanes = 4
        inbound-lanes = ${?REMOTE_INBOUND_LANES}

        # Number of outbound lanes for each outbound association. A value greater than 1
        # means that serialization and other work can be performed in parallel for different
        # destination actors.
        outbound-lanes = 1
        outbound-lanes = ${?REMOTE_OUTBOUND_LANES}

        # Size of the send queue for outgoing messages. Messages will be dropped if
        # the queue becomes full.
        outbound-message-queue-size = 3072
        outbound-message-queue-size = ${?REMOTE_OUTBOUND_MESSAGE_QUEUE_SIZE}
        # Size of the send queue for outgoing control messages, such as system messages.
        outbound-control-queue-size = 20000
        outbound-control-queue-size = ${?REMOTE_OUTBOUND_CONTROL_QUEUE_SIZE}
        # Size of the send queue for outgoing large messages. Messages will be dropped if
        # the queue becomes full.
        outbound-large-message-queue-size = 256
        outbound-large-message-queue-size = ${?REMOTE_OUTBOUND_LARGE_MESSAGE_QUEUE_SIZE}

        # Level of CPU time used, on a scale between 1 and 10, during backoff/idle.
        # The tradeoff is that to have low latency more CPU time must be used to be
        # able to react quickly on incoming messages or send as fast as possible after
        # backoff backpressure.
        # Level 1 strongly prefer low CPU consumption over low latency.
        # Level 10 strongly prefer low latency over low CPU consumption.
        idle-cpu-level = 1 # default: 5
        idle-cpu-level = ${?REMOTING_IDLE_CPU_LEVEL}
      }
    }
    watch-failure-detector.threshold = 12 # default 10
  }

  cluster {
    # Required for smooth rolling update in cluster sharding.
    app-version = ${ditto.version}

    # 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 = 60s

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

      # A new rebalance algorithm was included in Akka 2.6.10. It can reach optimal balance in
      # less rebalance rounds (typically 1 or 2 rounds)
      least-shard-allocation-strategy {
        # Maximum number of shards that will be rebalanced in one rebalance round.
        # The lower of this and `rebalance-relative-limit` will be used.
        # only when configured >0 (which is the default), this new shard allocation strategy is used
        rebalance-absolute-limit = 20
        rebalance-absolute-limit = ${?AKKA_CLUSTER_SHARDING_LEAST_SHARD_ALLOCATION_STRATEGY_REBALANCE_ABSOLUTE_LIMIT}

        # must be <=1.0: having configured e.g. 100 shards, a relative limit of 0.1 would result to "10"
        rebalance-relative-limit = 0.1
        rebalance-relative-limit = ${?AKKA_CLUSTER_SHARDING_LEAST_SHARD_ALLOCATION_STRATEGY_REBALANCE_RELATIVE_LIMIT}
      }
    }
  }

  coordinated-shutdown {
    phases {
      cluster-sharding-shutdown-region {
        timeout = 15s # default: 10s
        timeout = ${?AKKA_COORDINATED_SHUTDOWN_PHASES_CLUSTER_SHARDING_SHUTDOWN_REGION}
      }
      cluster-exiting {
        timeout = 15s # default: 10s
        timeout = ${?AKKA_COORDINATED_SHUTDOWN_PHASES_CLUSTER_EXITING_TIMEOUT}
      }
    }
  }
}

sharding-dispatcher {
  # Dispatcher is the name of the event-based dispatcher
  type = Dispatcher
  # What kind of ExecutionService to use
  executor = "org.eclipse.ditto.internal.utils.metrics.executor.InstrumentedForkJoinExecutorServiceConfigurator"
  # Configuration for the fork join pool
  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 = 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 = 5 # default is 5
  throughput = ${?SHARDING_DISPATCHER_THROUGHPUT}
}

akka.contrib.persistence.mongodb.mongo {
  driver = "akka.contrib.persistence.mongodb.ScalaDriverPersistenceExtension"

  # Write concerns are one of: Unacknowledged, Acknowledged, Journaled, ReplicaAcknowledged
  journal-write-concern = "Acknowledged" # By default was: "Journaled"
  journal-write-concern = ${?AKKA_PERSISTENCE_MONGO_JOURNAL_WRITE_CONCERN}
  journal-wtimeout = 10000
  journal-wtimeout = ${?AKKA_PERSISTENCE_MONGO_JOURNAL_WRITE_TIMEOUT}
  journal-fsync = false
  journal-fsync = ${?AKKA_PERSISTENCE_MONGO_JOURNAL_FSYNC}

  snaps-write-concern = "Acknowledged" # By default was: "Journaled"
  snaps-write-concern = ${?AKKA_PERSISTENCE_MONGO_SNAPS_WRITE_CONCERN}
  snaps-wtimeout = 5000
  snaps-wtimeout = ${?AKKA_PERSISTENCE_MONGO_SNAPS_WRITE_TIMEOUT}
  snaps-fsync = false
  snaps-fsync = ${?AKKA_PERSISTENCE_MONGO_SNAPS_FSYNC}

  realtime-enable-persistence = false

  metrics-builder {
    class = "org.eclipse.ditto.internal.utils.metrics.mongo.MongoMetricsBuilder"
    class = ${?MONGO_METRICS_BUILDER_CLASS}
  }
}