omearac/reactive-kafka-microservice-template

An example of a microservice template which uses akka/reactive-kafka to stream JSON messages

reactive-kafka-microservice-template's Issues

Clean up Actor System, Materializer and Execution Context in Traits

The way I'm currently passing the actor system around to the various actors and traits may not be the cleanest.

See how I've used the implicits in AkkaStream, ConsumerStream, ProducerStream and EventSourcing Traits and how there likely is a better way to do this.

Getting Invalid partition while running Main.scala

Hi,

Followed steps described in Link : https://github.com/omearac/reactive-kafka-microservice-template/blob/master/README.md to make kafka Up.

Getting below issue while running Main.scala

ERROR] a.a.RepointableActorRef: Error in stage [akka.kafka.internal.ProducerStage@9de0abb]: Invalid partition given with record: 4 is not in the range [0...0]. 
java.lang.IllegalArgumentException: Invalid partition given with record: 4 is not in the range [0...0].
	at org.apache.kafka.clients.producer.KafkaProducer.partition(KafkaProducer.java:717) ~[kafka-clients-0.10.0.1.jar:na]
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:459) ~[kafka-clients-0.10.0.1.jar:na]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:430) ~[kafka-clients-0.10.0.1.jar:na]
	at akka.kafka.internal.ProducerStage$$anon$1$$anon$3.onPush(ProducerStage.scala:67) ~[akka-stream-kafka_2.11-0.13.jar:0.13]
	at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:710) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:616) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:410) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.actor.Actor$class.aroundReceive(Actor.scala:496) ~[akka-actor_2.11-2.4.16.jar:na]
	at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529) ~[akka-stream_2.11-2.4.16.jar:na]
	at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526) ~[akka-actor_2.11-2.4.16.jar:na]
	at akka.actor.ActorCell.invoke(ActorCell.scala:495) ~[akka-actor_2.11-2.4.16.jar:na]
	at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257) ~[akka-actor_2.11-2.4.16.jar:na]
	at akka.dispatch.Mailbox.run(Mailbox.scala:224) ~[akka-actor_2.11-2.4.16.jar:na]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) ~[na:1.8.0_131]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) ~[na:1.8.0_131]
	at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_131]

microservices communicate with each other through event bus?

I have a question. I am confused with the comment in EventProducer.scala

• This actor receives local app events called "ExampleAppEvent"s which are initially published to the
• "internal" Akka Event Bus and which he is subscribed to. He then publishes the event messages to the Kafka
• topic called AppEventChannel. The idea would be that another microservice is subscribed to
• the AppEventChannel topic and can then react to events this microservice emits.

Does this mean a microservice communicates with another through both local event bus and Kafka?
If so, why do I need both?

Overuse of traits especially for declaring some default values. API clean.

Hello, I open this issue after seeing the details of the rest of the implementation. The previous issue is kind of related to this issue. The code should be refactored a bit. As this project is a template for future projects the high level of the structure should reflect that. I propose a structure like:

• kafka

• cordinators (or managers)

to separate the messaging layer from the light business logic. After that, inside Kafka there should be a high-level API for Consumer, Producer. Inside these objects should be placed the detailed API like Consumer.actor, Consumer.source, Producer.actor, Producer.sink, Flow, FlowACK. etc. All the deatils of the implementation should be placed in those classes and methods to abstract everything an give the user the ability to write simple code like:
val actorConsumer = Consumer.actor[String, String](topic: Topic, destination: ActorRef)
val source = Consumer.source[String, String](topic: Topic)