fix: failing the consumer if publishing without the destination is de…

…tected (#147) * fix: failing the consumer if publishing without the destination is detected * fixing impl
akka · Jan 14, 2025 · 19cf516 · 19cf516
1 parent d71aace
commit 19cf516
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 1 deletion.
diff --git a/akka-javasdk/src/main/scala/akka/javasdk/impl/SdkRunner.scala b/akka-javasdk/src/main/scala/akka/javasdk/impl/SdkRunner.scala
@@ -539,11 +539,13 @@ private final class Sdk(
       case clz if classOf[Consumer].isAssignableFrom(clz) =>
         val componentId = clz.getAnnotation(classOf[ComponentId]).value
         val consumerClass = clz.asInstanceOf[Class[Consumer]]
+        val consumerDest = consumerDestination(consumerClass)
         val consumerSpi =
           new ConsumerImpl[Consumer](
             componentId,
             () => wiredInstance(consumerClass)(sideEffectingComponentInjects(None)),
             consumerClass,
+            consumerDest,
             system.classicSystem,
             runtimeComponentClients.timerClient,
             sdkExecutionContext,

diff --git a/akka-javasdk/src/main/scala/akka/javasdk/impl/consumer/ConsumerImpl.scala b/akka-javasdk/src/main/scala/akka/javasdk/impl/consumer/ConsumerImpl.scala
@@ -5,9 +5,12 @@
 package akka.javasdk.impl.consumer
 
 import java.util.Optional
+
 import scala.concurrent.ExecutionContext
 import scala.concurrent.Future
 import scala.util.control.NonFatal
+
+import akka.Done
 import akka.actor.ActorSystem
 import akka.annotation.InternalApi
 import akka.javasdk.Metadata
@@ -31,6 +34,7 @@ import akka.javasdk.impl.telemetry.TraceInstrumentation
 import akka.javasdk.impl.timer.TimerSchedulerImpl
 import akka.javasdk.timer.TimerScheduler
 import akka.runtime.sdk.spi.BytesPayload
+import akka.runtime.sdk.spi.ConsumerDestination
 import akka.runtime.sdk.spi.SpiConsumer
 import akka.runtime.sdk.spi.SpiConsumer.Effect
 import akka.runtime.sdk.spi.SpiConsumer.Message
@@ -48,6 +52,7 @@ private[impl] final class ConsumerImpl[C <: Consumer](
     componentId: String,
     val factory: () => C,
     consumerClass: Class[C],
+    consumerDestination: Option[ConsumerDestination],
     _system: ActorSystem,
     timerClient: TimerClient,
     sdkExecutionContext: ExecutionContext,
@@ -96,11 +101,22 @@ private[impl] final class ConsumerImpl[C <: Consumer](
 
   private def toSpiEffect(message: Message, effect: Consumer.Effect): Future[Effect] = {
     effect match {
-      case ProduceEffect(msg, metadata) =>
+      case ProduceEffect(msg: Done, metadata) =>
         Future.successful(
           new SpiConsumer.ProduceEffect(
             payload = Some(serializer.toBytes(msg)),
             metadata = MetadataImpl.toSpi(metadata)))
+      case ProduceEffect(msg, metadata) =>
+        if (consumerDestination.isEmpty) {
+          val baseMsg = s"Consumer [$componentId] produced a message but no destination is defined."
+          log.error(baseMsg + " Add @Produce annotation or change the Consumer.Effect outcome.")
+          Future.successful(new SpiConsumer.ErrorEffect(new SpiConsumer.Error(baseMsg)))
+        } else {
+          Future.successful(
+            new SpiConsumer.ProduceEffect(
+              payload = Some(serializer.toBytes(msg)),
+              metadata = MetadataImpl.toSpi(metadata)))
+        }
       case AsyncEffect(futureEffect) =>
         futureEffect
           .flatMap { effect => toSpiEffect(message, effect) }