Several small things

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

@@ -110,7 +110,8 @@ object App extends OxApp, LazyLogging:
                             store,
                             clusterSettings,
                             kafkaConsumers,
-                            config.parallelism
+                            config.parallelism,
+                            config.domain.logThroughput
                         )
                     par(startHttpServer(), startAppWorkflow()).discard
             ExitCode.Success

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

@@ -44,14 +44,12 @@ object AppWorkflow:
         stateStore: StateStore,
         clusterSettings: Map[ClusterName, KafkaClusterSettings],
         kafkaConsumers: ConsumerMap,
-        parallelism: Parallelism
+        parallelism: Parallelism,
+        logDomainThroughput: Option[Boolean]
     )(using Ox): Unit =
-        DomainSources(channelConfigs, kafkaConsumers)
-            .buffer()
+        DomainSources(channelConfigs, kafkaConsumers, logDomainThroughput)
             .filterDomainMessages(channelConfigs, parallelism)
-            .buffer()
             .transformDomainMessageIds(channelConfigs)
-            .buffer()
             .transformDomainMessages(channelConfigs, parallelism)
             .buffer()
             .persistDomainMessages(stateStore)
@@ -65,13 +63,11 @@ object AppWorkflow:
             .inlineDomainMessages(relationConfigs, stateStore, parallelism)
             .buffer()
             .filterCodomainMessages(codomainConfig.filtering, parallelism)
-            .buffer()
             .transformCodomainMessages(codomainConfig.transformation, parallelism)
             .buffer()
             .persistCodomainMessages(stateStore, parallelism)
             .buffer()
             .propagateHeaders(codomainConfig.headerPropagationConfigs)
-            .buffer()
             .emitCodomainMessages(
                 KafkaSinkSettings(
                     codomainConfig.kafka,

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

@@ -19,21 +19,25 @@ object CodomainDeduplicationStage:
             configOpt: Option[CodomainDeduplicationConfig]
         ): Flow[(Seq[(QualifiedMessageId, Seq[MessageId])], Seq[Passthrough])] =
             val config = configOpt.getOrElse(defaultConfig)
-            in
-                .groupedWithin(config.batchSize, config.batchingDuration)
-                .map: batches =>
-                    val passthroughs: ListBuffer[Passthrough] = ListBuffer.empty
-                    val deduplicationMap: MutableMap[QualifiedMessageId, Set[MessageId]] = MutableMap.empty
-                    batches.foreach: batch =>
-                        batch._1 match
-                            case None =>
-                                ()
-                            case Some(domainMessageId, codomainMessageIds) =>
-                                deduplicationMap.updateWith(domainMessageId):
-                                    case None => Some(Set.empty ++ codomainMessageIds)
-                                    case Some(curCodomainMessageIds) =>
-                                        Some(curCodomainMessageIds ++ codomainMessageIds)
-                        passthroughs += batch._2
-                    (deduplicationMap.map((k, v) => (k, v.toSeq)).toSeq, passthroughs.sorted.toSeq)
+            if config.batchSize == 1 then
+                in.map: (payload, passthrough) =>
+                    (payload.toSeq.map(e => (e._1, e._2.toSeq)), Seq(passthrough))
+            else
+                in
+                    .groupedWithin(config.batchSize, config.batchingDuration)
+                    .map: batches =>
+                        val passthroughs: ListBuffer[Passthrough] = ListBuffer.empty
+                        val deduplicationMap: MutableMap[QualifiedMessageId, Set[MessageId]] = MutableMap.empty
+                        batches.foreach: batch =>
+                            batch._1 match
+                                case None =>
+                                    ()
+                                case Some(domainMessageId, codomainMessageIds) =>
+                                    deduplicationMap.updateWith(domainMessageId):
+                                        case None => Some(Set.empty ++ codomainMessageIds)
+                                        case Some(curCodomainMessageIds) =>
+                                            Some(curCodomainMessageIds ++ codomainMessageIds)
+                            passthroughs += batch._2
+                        (deduplicationMap.map((k, v) => (k, v.toSeq)).toSeq, passthroughs.sorted.toSeq)
 
 case class CodomainDeduplicationConfig(batchSize: Int, batchingDuration: FiniteDuration) derives ConfigReader

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

@@ -14,14 +14,16 @@ object DomainSources:
 
     def apply(
         configs: ChannelConfigs,
-        consumers: ConsumerMap
+        consumers: ConsumerMap,
+        logThroughput: Option[Boolean]
     )(using Ox): Flow[(Option[(QualifiedMessageId, Option[Message])], Passthrough)] =
         configs.channels
             .map: config =>
                 // for now we go with consumer name == domain name
                 val consumerName = ConsumerName(config.name.toString)
-                DomainSource(
+                val src = DomainSource(
                     config,
                     KafkaSourceSettings(config.kafka, consumerName, consumers(consumerName))
-                ).count(s"domain source ${config.name.toString}")
+                )
+                if logThroughput.getOrElse(false) then src.count(s"domain source ${config.name.toString}") else src
             .mergeFair()

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

@@ -97,8 +97,8 @@ object KafkaSink extends LazyLogging:
                         producerRecords.foreach(publishChannel.send)
                         offsets.foreach(committerChannel.send)
                     .mapStateful(0): (cnt, _) =>
-                        if cnt % 100 == 0 then
-                            logger.info("Published and committed 100 batches of codomain messages to Kafka")
+                        if cnt % 1000 == 0 then
+                            logger.info("Published and committed 1000 batches of codomain messages to Kafka")
                         (cnt + 1, ())
                     .runDrain()
 

src/main/scala/de/otto/anthology/config/DomainConfig.scala

@@ -3,5 +3,6 @@ package de.otto.anthology.config
 import de.otto.anthology.ChannelName
 import pureconfig.ConfigReader
 
-case class DomainConfig(channels: Seq[ChannelConfig], relations: Seq[RelationConfig]) derives ConfigReader:
+case class DomainConfig(channels: Seq[ChannelConfig], relations: Seq[RelationConfig], logThroughput: Option[Boolean])
+    derives ConfigReader:
     val channelsByName: Map[ChannelName, ChannelConfig] = channels.map(c => (c.name, c)).toMap