1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.technical
21 
22 import akka.actor.typed.eventstream.EventStream.Subscribe
23 import akka.actor.typed.scaladsl.Behaviors
24 import akka.actor.typed.{Behavior, LogOptions, PostStop}
25 import com.sksamuel.avro4s.{Encoder, SchemaFor}
26 import org.apache.kafka.clients.producer._
27 import org.apache.kafka.common.serialization.{Serializer, StringSerializer}
28 import org.make.api.extensions.KafkaConfiguration
29 import org.make.core.{AvroSerializers, WithEventId}
30 import org.slf4j.event.Level
31 
32 import scala.concurrent.duration.DurationInt
33 import java.util.{Properties, UUID}
34 import scala.reflect.ClassTag
35 
36 abstract class KafkaProducerBehavior[Event: ClassTag, Wrapper: SchemaFor: Encoder] extends AvroSerializers {
37 
38   protected val topicKey: String
39 
40   protected def createProducerConfiguration(kafkaConfiguration: KafkaConfiguration, name: String): Properties = {
41     val props = new Properties()
42     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.connectionString)
43     props.put(ProducerConfig.ACKS_CONFIG, "all")
44     props.put(ProducerConfig.RETRIES_CONFIG, "3")
45     props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384")
46     props.put(ProducerConfig.LINGER_MS_CONFIG, "1")
47     props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.make.api.technical.MakePartitioner")
48     props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432")
49     props.put(ProducerConfig.CLIENT_ID_CONFIG, name)
50     props
51   }
52 
53   protected def createProducer(kafkaConfiguration: KafkaConfiguration, name: String): KafkaProducer[String, Wrapper] = {
54     val props: Properties = createProducerConfiguration(kafkaConfiguration, name)
55     val valueSerializer: Serializer[Wrapper] =
56       new MakeKafkaAvroSerializer[Wrapper](kafkaConfiguration.schemaRegistry)
57     new KafkaProducer(props, new StringSerializer(), valueSerializer)
58   }
59 
60   protected def wrapEvent(event: Event): Wrapper
61 
62   @SuppressWarnings(Array("org.wartremover.warts.Null"))
63   def createBehavior(name: String): Behavior[Event] = {
64     Behaviors.setup { context =>
65       val kafkaConfiguration = KafkaConfiguration(context.system)
66       val topic = kafkaConfiguration.topic(topicKey)
67       val producer = createProducer(kafkaConfiguration, name)
68       context.system.eventStream ! Subscribe(context.self)
69 
70       val sendCallBack: Callback = (r: RecordMetadata, e: Exception) => {
71         val topic = Option(r).map(_.topic()).getOrElse("unknown")
72         Option(e).foreach(e => context.log.error(s"Error when producing message on topic $topic", e))
73       }
74       context.log.info(s"Starting producer for topic $topic")
75       Behaviors.logMessages[Event](
76         LogOptions().withLevel(Level.DEBUG),
77         Behaviors
78           .receiveMessage[Event] { event =>
79             if (producer.partitionsFor(topic).size > 0) {
80               val record = wrapEvent(event)
81               producer.send(
82                 new ProducerRecord[String, Wrapper](
83                   topic,
84                   None.orNull,
85                   System.currentTimeMillis(),
86                   extractKey(record).getOrElse(UUID.randomUUID().toString),
87                   record
88                 ),
89                 sendCallBack
90               )
91               Behaviors.same
92             } else {
93               context.scheduleOnce(100.milliseconds, context.self, event)
94               Behaviors.same
95             }
96           }
97           .receiveSignal {
98             case (_, PostStop) =>
99               context.log.info(s"Stopping producer for topic $topic")
100               producer.close()
101               Behaviors.same
102           }
103       )
104     }
105   }
106 
107   private def extractKey(message: Wrapper): Option[String] = {
108     message match {
109       case event: WithEventId => event.eventId.map(_.value)
110       case _                  => None
111     }
112   }
113 }