Rework config

Dockerfile

@@ -31,8 +31,8 @@ COPY --from=build --chown=anthology:anthology /build/build/install/anthology/ .
 #   - application.yaml
 #   - any transform JSON files referenced in application.yaml
 #
-#  Pass ANTHOLOGY_CREDENTIALS at runtime:
-#   docker run -e ANTHOLOGY_CREDENTIALS='{"cluster-a":{"username":"...","password":"..."}}' ...
+#  Pass ANTHOLOGY_ADDITIONAL_KAFKA_PROPERTIES at runtime:
+#   docker run -e ANTHOLOGY_ADDITIONAL_KAFKA_PROPERTIES='{"cluster-a":{"username":"...","password":"..."}}' ...
 COPY --chown=anthology:anthology config/ /config/
 
 ENV ANTHOLOGY_CONFIG_FILE=/config/application.yaml

src/main/scala/de/otto/anthology/App.scala

@@ -1,20 +1,20 @@
 package de.otto.anthology
 
 import com.typesafe.scalalogging.LazyLogging
+import de.otto.anthology.config.AdditionalKafkaPropertiesLoader
 import de.otto.anthology.config.AnthologyConfig
 import de.otto.anthology.config.AnthologyConfigFactory
+import de.otto.anthology.config.ChannelConfigs
 import de.otto.anthology.config.CliConf
 import de.otto.anthology.config.CodomainConfig
-import de.otto.anthology.config.CredentialsLoader
-import de.otto.anthology.config.DomainConfigs
-import de.otto.anthology.config.DomainRelationConfigs
 import de.otto.anthology.config.KafkaClusterSettings
+import de.otto.anthology.config.RelationConfigs
 import de.otto.anthology.http.Server
-import de.otto.anthology.kafka.AggregateDeserializer
-import de.otto.anthology.kafka.AggregateIdDeserializer
 import de.otto.anthology.kafka.ClusterName
 import de.otto.anthology.kafka.ConsumerMap
 import de.otto.anthology.kafka.ConsumerName
+import de.otto.anthology.kafka.MessageDeserializer
+import de.otto.anthology.kafka.MessageIdDeserializer
 import de.otto.anthology.statestore.RocksDBConfig
 import de.otto.anthology.statestore.RocksDBStateStore
 import de.otto.anthology.statestore.StateStore
@@ -44,15 +44,17 @@ object App extends OxApp, LazyLogging:
                     val config: AnthologyConfig = AnthologyConfigFactory(cliConfig.anthologyConfigFile.toOption)
                     logger.info(s"Starting ${config.name}...")
 
-                    val credentials: Map[ClusterName, Map[String, String]] =
-                        CredentialsLoader(cliConfig.anthologyCredentials.toOption)
-                    logger.info("Credentials loaded successfully")
+                    val additionalKafkaProps: Map[ClusterName, Map[String, String]] =
+                        AdditionalKafkaPropertiesLoader(cliConfig.anthologyAdditionalKafkaProperties.toOption)
+                    logger.info("Additional Kafka properties loaded successfully")
 
                     val clusterSettings: Map[ClusterName, KafkaClusterSettings] =
-                        config.kafkaClusters.map(cc => cc.name -> KafkaClusterSettings(cc, credentials(cc.name))).toMap
+                        config.kafkaClusters
+                            .map(cc => cc.name -> KafkaClusterSettings(cc, additionalKafkaProps(cc.name)))
+                            .toMap
 
-                    val domainConfigs: DomainConfigs = DomainConfigs(config.domains)
-                    val domainRelationConfigs: DomainRelationConfigs = DomainRelationConfigs(config.domainRelations)
+                    val channelConfigs: ChannelConfigs = ChannelConfigs(config.domain.channels)
+                    val relationConfigs: RelationConfigs = RelationConfigs(config.domain.relations)
                     val codomainConfig: CodomainConfig = config.codomain
                     logger.info("Domain and codomain settings initialized successfully")
 
@@ -64,19 +66,19 @@ object App extends OxApp, LazyLogging:
                     logger.info("State store initialized successfully")
 
                     val kafkaConsumers: ConsumerMap =
-                        domainConfigs.domains
+                        channelConfigs.channels
                             .map: dConfig =>
                                 val cluster = clusterSettings(dConfig.kafka.cluster)
-                                val creds = cluster.credentials
-                                val baseSettings: ConsumerSettings[AggregateId, Option[Aggregate]] =
+                                val additionalProps = cluster.additionalProperties
+                                val baseSettings: ConsumerSettings[MessageId, Option[Message]] =
                                     ConsumerSettings
                                         .default(dConfig.kafka.consumerGroup)
                                         .bootstrapServers(cluster.config.bootstrapServers.split(",").map(_.trim)*)
-                                        .keyDeserializer(AggregateIdDeserializer)
-                                        .valueDeserializer(AggregateDeserializer)
+                                        .keyDeserializer(MessageIdDeserializer)
+                                        .valueDeserializer(MessageDeserializer)
                                         .autoOffsetReset(AutoOffsetReset.Earliest)
-                                val consumerSettings: ConsumerSettings[AggregateId, Option[Aggregate]] =
-                                    creds.foldLeft(baseSettings)((s, k2v) => s.property(k2v._1, k2v._2))
+                                val consumerSettings: ConsumerSettings[MessageId, Option[Message]] =
+                                    additionalProps.foldLeft(baseSettings)((s, k2v) => s.property(k2v._1, k2v._2))
                                 // for now, we go with consumer name == domain name
                                 ConsumerName(dConfig.name.toString) -> consumerSettings.toThreadSafeConsumerWrapper
                             .toMap
@@ -87,8 +89,8 @@ object App extends OxApp, LazyLogging:
                     def startHttpServer: Unit = Server().start()
                     def startAppWorkflow: Unit =
                         AppWorkflow.run(
-                            domainConfigs,
-                            domainRelationConfigs,
+                            channelConfigs,
+                            relationConfigs,
                             codomainConfig,
                             store,
                             clusterSettings,

src/main/scala/de/otto/anthology/AppWorkflow.scala

@@ -1,26 +1,26 @@
 package de.otto.anthology
 
-import de.otto.anthology.CodomainCompositionStage.composeCodomainAggregates
-import de.otto.anthology.CodomainDeduplicationStage.deduplicateCodomainAggregates
-import de.otto.anthology.CodomainInliningStage.inlineDomainAggregates
-import de.otto.anthology.CodomainPersistenceStage.persistCodomainAggregates
-import de.otto.anthology.CodomainTriggeringStage.triggerAffectedCodomainAggregates
-import de.otto.anthology.DomainLinkingStage.linkDomainAggregates
-import de.otto.anthology.DomainPersistenceStage.persistDomainAggregates
-import de.otto.anthology.KafkaSink.emit
+import de.otto.anthology.CodomainCompositionStage.composeCodomainMessages
+import de.otto.anthology.CodomainDeduplicationStage.deduplicateCodomainMessages
+import de.otto.anthology.CodomainInliningStage.inlineDomainMessages
+import de.otto.anthology.CodomainPersistenceStage.persistCodomainMessages
+import de.otto.anthology.CodomainTriggeringStage.triggerAffectedCodomainMessages
+import de.otto.anthology.DomainLinkingStage.linkDomainMessages
+import de.otto.anthology.DomainPersistenceStage.persistDomainMessages
+import de.otto.anthology.KafkaSink.emitCodomainMessages
+import de.otto.anthology.config.ChannelConfigs
 import de.otto.anthology.config.CodomainConfig
-import de.otto.anthology.config.DomainConfigs
-import de.otto.anthology.config.DomainRelationConfigs
 import de.otto.anthology.config.KafkaClusterSettings
-import de.otto.anthology.filtering.CodomainFilteringStage.filterCodomainAggregates
-import de.otto.anthology.filtering.DomainFilteringStage.filterDomainAggregates
+import de.otto.anthology.config.RelationConfigs
+import de.otto.anthology.filtering.CodomainFilteringStage.filterCodomainMessages
+import de.otto.anthology.filtering.DomainFilteringStage.filterDomainMessages
 import de.otto.anthology.headerpropagation.HeaderPropagationStage.propagateHeaders
 import de.otto.anthology.kafka.ClusterName
 import de.otto.anthology.kafka.ConsumerMap
 import de.otto.anthology.statestore.StateStore
-import de.otto.anthology.transformation.CodomainTransformationStage.transformCodomainAggregates
-import de.otto.anthology.transformation.DomainTransformationStage.transformDomainAggregateIds
-import de.otto.anthology.transformation.DomainTransformationStage.transformDomainAggregates
+import de.otto.anthology.transformation.CodomainTransformationStage.transformCodomainMessages
+import de.otto.anthology.transformation.DomainTransformationStage.transformDomainMessageIds
+import de.otto.anthology.transformation.DomainTransformationStage.transformDomainMessages
 import ox.Ox
 import ox.channels.BufferCapacity
 
@@ -30,51 +30,51 @@ object AppWorkflow:
 
     /** Sets up and runs Anthology's main application workflow.
       *
-      * @param domainConfigs
-      *   Configuration relating to the domains.
-      * @param domainRelationConfigs
-      *   Configuration relating to the relations between the domains.
+      * @param channelConfigs
+      *   Configuration relating to the channels.
+      * @param relationConfigs
+      *   Configuration relating to the relations between the messages.
       * @param codomainConfig
       *   Codomain-related configuration.
       * @param stateStore
       *   A fully configured instance of the state store.
       */
     def run(
-        domainConfigs: DomainConfigs,
-        domainRelationConfigs: DomainRelationConfigs,
+        channelConfigs: ChannelConfigs,
+        relationConfigs: RelationConfigs,
         codomainConfig: CodomainConfig,
         stateStore: StateStore,
         clusterSettings: Map[ClusterName, KafkaClusterSettings],
         kafkaConsumers: ConsumerMap,
         parallelism: Parallelism
     )(using Ox): Unit =
-        DomainSources(domainConfigs, kafkaConsumers)
+        DomainSources(channelConfigs, kafkaConsumers)
             .buffer()
-            .filterDomainAggregates(domainConfigs, parallelism)
+            .filterDomainMessages(channelConfigs, parallelism)
             .buffer()
-            .transformDomainAggregateIds(domainConfigs)
+            .transformDomainMessageIds(channelConfigs)
             .buffer()
-            .transformDomainAggregates(domainConfigs, parallelism)
+            .transformDomainMessages(channelConfigs, parallelism)
             .buffer()
-            .persistDomainAggregates(stateStore)
+            .persistDomainMessages(stateStore)
             .buffer()
-            .linkDomainAggregates(domainRelationConfigs, stateStore, parallelism)
+            .linkDomainMessages(relationConfigs, stateStore, parallelism)
             .buffer()
-            .triggerAffectedCodomainAggregates(domainRelationConfigs, stateStore, parallelism)
-            .deduplicateCodomainAggregates(codomainConfig.deduplication)
-            .composeCodomainAggregates(stateStore)
+            .triggerAffectedCodomainMessages(relationConfigs, stateStore, parallelism)
+            .deduplicateCodomainMessages(codomainConfig.deduplication)
+            .composeCodomainMessages(stateStore)
             .buffer()
-            .inlineDomainAggregates(domainRelationConfigs, stateStore, parallelism)
+            .inlineDomainMessages(relationConfigs, stateStore, parallelism)
             .buffer()
-            .filterCodomainAggregates(codomainConfig.filtering, parallelism)
+            .filterCodomainMessages(codomainConfig.filtering, parallelism)
             .buffer()
-            .transformCodomainAggregates(codomainConfig.transformation, parallelism)
-            .buffer()
-            .persistCodomainAggregates(stateStore, parallelism)
+            .transformCodomainMessages(codomainConfig.transformation, parallelism)
             .buffer()
+            .persistCodomainMessages(stateStore, parallelism)
             .buffer()
             .propagateHeaders(codomainConfig.headerPropagationConfigs)
-            .emit(
+            .buffer()
+            .emitCodomainMessages(
                 KafkaSinkSettings(
                     codomainConfig.kafka,
                     clusterSettings(codomainConfig.kafka.cluster),

src/main/scala/de/otto/anthology/CodomainCompositionStage.scala

@@ -3,11 +3,11 @@ package de.otto.anthology
 import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.node.ObjectNode
 import com.typesafe.scalalogging.LazyLogging
-import de.otto.anthology.AggregateId
-import de.otto.anthology.AggregateName
-import de.otto.anthology.DomainName
+import de.otto.anthology.ChannelName
 import de.otto.anthology.JsonSupport.mapper
-import de.otto.anthology.QualifiedAggregateId
+import de.otto.anthology.MessageFormatName
+import de.otto.anthology.MessageId
+import de.otto.anthology.QualifiedMessageId
 import de.otto.anthology.kafka.Passthrough
 import de.otto.anthology.statestore.StateStore
 import de.otto.anthology.statestore.StateStoreSection
@@ -18,61 +18,61 @@ import scala.util.control.NonFatal
 
 object CodomainCompositionStage extends LazyLogging:
 
-    extension (in: Flow[(Seq[(QualifiedAggregateId, Seq[AggregateId])], Seq[Passthrough])])
-        def composeCodomainAggregates(stateStore: StateStore): Flow[(Seq[AggregateId], Seq[Passthrough])] =
+    extension (in: Flow[(Seq[(QualifiedMessageId, Seq[MessageId])], Seq[Passthrough])])
+        def composeCodomainMessages(stateStore: StateStore): Flow[(Seq[MessageId], Seq[Passthrough])] =
             in.map: (payloads, passthroughs) =>
-                val payloadsOut: Seq[AggregateId] =
-                    payloads.flatMap: (qaid, codomainAggregateIds) =>
-                        codomainAggregateIds.flatMap: codomainAggregateId =>
+                val payloadsOut: Seq[MessageId] =
+                    payloads.flatMap: (qmid, codomainMessageIds) =>
+                        codomainMessageIds.flatMap: codomainMessageId =>
                             try
-                                val codomainKey = s"${StateStoreSection.STA}/$codomainAggregateId"
-                                val codomainAggregate: ObjectNode =
+                                val codomainKey = s"${StateStoreSection.STA}/$codomainMessageId"
+                                val codomainMessage: ObjectNode =
                                     stateStore
                                         .getJson(codomainKey)
                                         .fold(mapper.createObjectNode())(_.asInstanceOf[ObjectNode])
-                                compose(qaid, codomainAggregate, stateStore)
-                                if codomainAggregate.isEmpty then stateStore.delete(codomainKey)
-                                else stateStore.putJson(codomainKey, codomainAggregate)
-                                Some(codomainAggregateId)
+                                compose(qmid, codomainMessage, stateStore)
+                                if codomainMessage.isEmpty then stateStore.delete(codomainKey)
+                                else stateStore.putJson(codomainKey, codomainMessage)
+                                Some(codomainMessageId)
                             catch
                                 case NonFatal(ex) =>
                                     logger.error(
-                                        s"Error processing domain aggregate ($qaid) and codomain aggregate ($codomainAggregateId): ${ex.stackTraceAsString}"
+                                        s"Error processing domain message ($qmid) and codomain message ($codomainMessageId): ${ex.stackTraceAsString}"
                                     )
                                     None
                 (payloadsOut.distinct, passthroughs)
 
     private def compose(
-        currentDomainAggregateId: QualifiedAggregateId,
-        codomainAggregate: ObjectNode,
+        currentDomainMessageId: QualifiedMessageId,
+        codomainMessage: ObjectNode,
         stateStore: StateStore
     ): Unit =
-        val currentDomainAggregateOpt: Option[JsonNode] =
-            stateStore.getJson(s"${StateStoreSection.DOM}/$currentDomainAggregateId")
-        val currentDomainAggregateMap: ObjectNode =
-            Option(codomainAggregate.get(currentDomainAggregateId.qualifierString))
+        val currentDomainMessageOpt: Option[JsonNode] =
+            stateStore.getJson(s"${StateStoreSection.DOM}/$currentDomainMessageId")
+        val currentDomainMessageMap: ObjectNode =
+            Option(codomainMessage.get(currentDomainMessageId.qualifierString))
                 .fold(mapper.createObjectNode())(_.asInstanceOf[ObjectNode])
-        setObject(currentDomainAggregateMap, currentDomainAggregateId.id.toString, currentDomainAggregateOpt)
+        setObject(currentDomainMessageMap, currentDomainMessageId.id.toString, currentDomainMessageOpt)
         setObject(
-            codomainAggregate,
-            currentDomainAggregateId.qualifierString,
-            if currentDomainAggregateMap.isEmpty then None else Some(currentDomainAggregateMap)
+            codomainMessage,
+            currentDomainMessageId.qualifierString,
+            if currentDomainMessageMap.isEmpty then None else Some(currentDomainMessageMap)
         )
 
-        val next: Set[QualifiedAggregateId] =
+        val next: Set[QualifiedMessageId] =
             stateStore
                 .getStringSet(
-                    s"${StateStoreSection.LNK}/$currentDomainAggregateId"
+                    s"${StateStoreSection.LNK}/$currentDomainMessageId"
                 )
                 .map: entry =>
                     val splittedEntry = entry.split("/")
-                    QualifiedAggregateId(
-                        DomainName(splittedEntry(0)),
-                        AggregateName(splittedEntry(1)),
-                        AggregateId(splittedEntry(2))
+                    QualifiedMessageId(
+                        ChannelName(splittedEntry(0)),
+                        MessageFormatName(splittedEntry(1)),
+                        MessageId(splittedEntry(2))
                     )
-        next.foreach: nextAggregateId =>
-            compose(nextAggregateId, codomainAggregate, stateStore)
+        next.foreach: nextMessageId =>
+            compose(nextMessageId, codomainMessage, stateStore)
 
     private def setObject(parent: ObjectNode, name: String, objOpt: Option[JsonNode]): Unit =
         objOpt match