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

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

    kubernetes-api {
      pod-namespace = "default" // in which namespace cluster is running
      pod-namespace = ${?K8S_NAMESPACE}

      pod-label-selector = "actorSystemName=ditto-cluster"
      pod-label-selector = ${?K8S_LABEL_SELECTOR}

      pod-port-name = "akka-mgmt-http" // name of cluster management port
      pod-port-name = ${?K8S_MANAGEMENT_PORT}
    }
  }

  coordinated-shutdown.exit-jvm=on

  management {
    cluster.bootstrap {
      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}

        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.services.utils.cluster.JsonifiableSerializer"
    }

    # 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.model.base.json.Jsonifiable" = json
      "org.eclipse.ditto.model.base.exceptions.DittoRuntimeException" = json
    }

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

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

  remote {
    log-remote-lifecycle-events = on

    netty.tcp {
      # InetAddress.getLocalHost.getHostAddress is used if empty
      hostname = ""
      hostname = ${?REMOTE_HOSTNAME}
      port = 2551
      port = ${?REMOTE_PORT}

      bind-hostname = ${?BIND_HOSTNAME}
      bind-port = ${?BIND_TCP_PORT}

      # maximum-frame-size = 128000b # this is the default
      maximum-frame-size = 256000b # 256 KiB
      maximum-frame-size = ${?REMOTE_MAX_FRAMESIZE}
      # send-buffer-size = 256000b # this is the default
      send-buffer-size = 320000b # 320 KiB
      send-buffer-size = ${?REMOTE_SEND_BUFFERSIZE}
      # receive-buffer-size = 256000b # this is the default
      receive-buffer-size = 320000b # 320 KiB
      receive-buffer-size = ${?REMOTE_RECEIVE_BUFFERSIZE}
    }

    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

        # Size of the send queue for outgoing messages. Messages will be dropped if
        # the queue becomes full.
        outbound-message-queue-size = 3072
        # Size of the send queue for outgoing control messages, such as system messages.
        outbound-control-queue-size = 3072
        # Size of the send queue for outgoing large messages. Messages will be dropped if
        # the queue becomes full.
        outbound-large-message-queue-size = 256

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

sharding-dispatcher {
  # Dispatcher is the name of the event-based dispatcher
  type = Dispatcher
  # What kind of ExecutionService to use
  executor = "fork-join-executor"
  # Configuration for the fork join pool
  fork-join-executor {
    # Min number of threads to cap factor-based parallelism number to
    parallelism-min = 8
    # Parallelism (threads) ... ceil(available processors * factor)
    parallelism-factor = 3.0
    # Max number of threads to cap factor-based parallelism number to
    parallelism-max = 128
  }
  # 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
}

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

  # Write concerns are one of: ErrorsIgnored, Unacknowledged, Acknowledged, Journaled, ReplicaAcknowledged
  journal-write-concern = "Acknowledged" # By default was: "Journaled"
  journal-wtimeout = 10000
  journal-fsync = false

  snaps-write-concern = "Acknowledged" # By default was: "Journaled"
  snaps-wtimeout = 5000
  snaps-fsync = false

  realtime-enable-persistence = false
}