Schedule messages #2

lacarvalho91 · 2017-08-07T09:57:38Z

Will also be looking into adding metrics but is ready for an initial review.

I didn't realise git pair would be amending commit author outside of using git pair-commit.

Fix error that occasionally happens when generating random Schedule
Memory leak in tests

…r stream error handling. Metrics?

… have been emitted. Tests passing. TODO: error handling in ScheduleReader stream.

…implicit Sink[ErrorType, _].

oliverlockwood · 2017-08-07T10:05:48Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/SchedulerApp.scala


-object SchedulerApp extends App with AkkaComponents with LazyLogging {
+object SchedulerApp extends App with LazyLogging with AkkaComponents {


For my own understanding - is there any logical reason why the order of these is swapped? I had no awareness that this made any difference to the compiler.

I believe there are circumstances where it would make a difference, but in this case - no.

Spent some time trying to work out how to properly inject the implicits from AkkaComponents so it would've moved during that experimentation.

lacarvalho91 · 2017-08-07T16:19:47Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/AkkaComponents.scala


  implicit val system = ActorSystem("kafka-message-scheduler")
-  implicit val materializer = ActorMaterializer()
+
+  val decider: Supervision.Decider = { t =>


this stuff and the tests for it are a WIP

paoloambrosio-skyuk

Partial review

paoloambrosio-skyuk · 2017-08-07T15:39:43Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/SchedulerApp.scala


  sys.addShutdownHook {
    logger.info("Kafka Message Scheduler shutting down...")
-    Await.ready(runningStream.shutdown(), conf.shutdownTimeout.stream)
-    Await.ready(system.terminate(), conf.shutdownTimeout.system)
+    Rewriter.stop(app).value


All of this is a bit messy Grafter stuff mixed with Akka, Kamon, ... all working in a different way. It would be good to create an abstraction so that we can close everything in the same way.

Seconded, it's pretty hard to follow everything at the moment.

agreed, will take some time to come up with that i would have thought. for this PR or another?

paoloambrosio-skyuk · 2017-08-07T15:45:12Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/SchedulingActor.scala

+          log.info(s"Updating schedule $scheduleId")
+        else
+          log.info(s"Creating schedule $scheduleId")
+        val cancellable = scheduler.scheduleOnce(timeFromNow(schedule.time)){


I usually prefer to schedule a message to the actor itself instead of executing a function, as it makes it easier to handle concurrency properly.

paoloambrosio-skyuk · 2017-08-07T15:52:16Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/domain/ApplicationError.scala

@@ -22,4 +28,8 @@ object ApplicationError {
    case schemaError: InvalidSchemaError => invalidSchemaErrorShow.show(schemaError)
    case messageFormatError: AvroMessageFormatError => avroMessageFormatErrorShow.show(messageFormatError)
  }
+
+  //TODO: is this enough for now?


Can we use a Reader? Happy to postpone it to later. It should really be created from the raw configuration (not application config). We should later create a project with all our common code (a la akka-stream-contrib).

paoloambrosio-skyuk · 2017-08-07T16:20:23Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/domain/ScheduleData.scala

+
+  case class Schedule(time: OffsetDateTime, topic: String, key: Array[Byte], value: Array[Byte]) extends ScheduleData
+
+  case class ScheduleMetadata(scheduleId: ScheduleId, topic: String) extends ScheduleData


Something like ScheduledMessage(topic: String, key: Array[Byte], value: Array[Byte]) and ScheduleDeletion(scheduleId: ScheduleId) might be more appropriate (note the topic missing from ScheduleDeletion as the ...ToProducerRecord should take it from config.

paoloambrosio-skyuk · 2017-08-08T09:59:48Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/package.scala


 package object scheduler extends LazyLogging {

-  case class AppConfig(scheduler: SchedulerConfig)
+  type DecodeResult = Either[ApplicationError, (ScheduleId, Option[Schedule])]


Not sure DecodeResult is the right abstraction. It does not communicate what it is. With that name I would have expected type DecodeResult[T] = Either[DecodingError, T] where DecodingError inherits from ApplicationError... but that might not deserve being a separate type.

will remove type alias

oliverlockwood · 2017-08-08T10:05:46Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/streams/ScheduleReader.scala

+  def reader(implicit system: ActorSystem, materializer: ActorMaterializer): Reader[AppConfig, ScheduleReader] =
+    for {
+      config <- SchedulerConfig.reader
+      schedulingActorRef <- SchedulingActor.reader


Do I take it this is all about using Reader for dependency injection, as per http://eed3si9n.com/herding-cats/Reader.html?

paoloambrosio-skyuk · 2017-08-08T10:11:56Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/streams/EitherFanOut.scala

+import akka.stream.stage.{GraphStage, InHandler, OutHandler}
+
+//Taken from: https://stackoverflow.com/a/38445121/8424807
+class EitherFanOut[L, R] extends GraphStage[FanOutShape2[Either[L, R], L, R]] {


Isn't PartitionWith basically what this is?

I looked at that but maybe I misunderstood it - will take another look.

it works! 🎉

Suggest that to the StackOverflow question 😄

paoloambrosio-skyuk · 2017-08-08T10:20:51Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/streams/ScheduleReader.scala

+import com.sky.kafka.message.scheduler.kafka._
+import com.typesafe.scalalogging.LazyLogging
+
+case class ScheduleReader(config: SchedulerConfig, scheduleSource: Source[DecodeResult, Control], schedulingSink: Sink[Any, NotUsed])


What about adding Scaladoc about for the purpose of these classes? In this case this would be something like...

/** * Provides stream from the schedule source to the scheduling actor. */

Reading through the previous class PartitionedSink I was thinking exactly the same thing! Possibly worth pairing on the docs with someone?

Also, I'm not sure about the name ScheduleReader. Maybe I'm being dense but I got confused between this and the Reader concept that's introduced in this PR.

How about ScheduleInputHandler or something else?

ScheduleReaderStream?

I'm not a fan of changing it, because ScheduleReader describes it very well IMO. Does the above distinguish it enough?

that ruins it actually as it would be SchedulerReaderStream.stream.

I would rather it stayed ScheduleReader, perhaps we should discuss as a team

paoloambrosio-skyuk · 2017-08-08T10:30:29Z

scheduler/src/test/scala/com/sky/kafka/message/scheduler/AkkaComponentsSpec.scala

+class AkkaComponentsSpec extends AkkaBaseSpec {
+
+  "decider" should {
+    "return a 'Restart' supervision strategy" in {


Hopefully there is a higher level test that shows the effect of this error handling logic. Also not sure of the value that this test is giving. It looks to me like testing accessors (testing something that is declarative).

this will evolve into a test that is testing we report metrics in the supervisor strategy, its a WIP

paoloambrosio-skyuk · 2017-08-08T10:32:04Z

scheduler/src/test/scala/com/sky/kafka/message/scheduler/SchedulingActorSpec.scala

+      verify(mockSourceQueue).offer((scheduleId, updatedSchedule))
+    }
+
+    "send an Ack to the sender when receiving an Init message" in  new SchedulingActorTest {


Shouldn't we send the Ack back on every scheduling message as well as on Init? Yes, just seen that createSchedule is masking what is happening. I think in this case we want to see both the message and be told we expect the Ack.

At this point we can also rename this test case as "does nothing on init" or something on those lines, given that we don't talk about messages in the other test names.

we already see both the message and that we expect the Ack - unless I'm misunderstanding?

have renamed the test

paoloambrosio-skyuk · 2017-08-08T10:36:00Z

scheduler/src/test/scala/common/EmbeddedKafka.scala


 object EmbeddedKafka {

  val kafkaServer = new KafkaServer

  val bootstrapServer = s"localhost:${kafkaServer.kafkaPort}"

+  implicit class KafkaServerOps(val kafkaServer: KafkaServer) extends AnyVal {


We definitely need to add a comment about why this is here, where it comes from, and why we haven't fixed code quality issues.

oliverlockwood · 2017-08-08T10:37:54Z

High-level: I'm not convinced by the root package being com.sky.kafka.message.scheduler. It leads to subpackages like com.sky.kafka.message.scheduler.kafka which seems weird to me.

How about the root package being com.sky.kafkascheduler?

oliverlockwood · 2017-08-08T10:40:09Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/kafka/KafkaStream.scala

+object KafkaStream {
+
+  def source[T](config: SchedulerConfig)(implicit system: ActorSystem, crDecoder: ConsumerRecordDecoder[T]): Source[T, Control] =
+    Consumer.plainSource(


Do we need to add a Github issue to track moving to a committableSource to provide at-least-once semantics?

Also in the meantime for any commits to be made, you'll need to change enable.auto.commit to true in application.conf.

oliverlockwood · 2017-08-08T10:50:52Z

From avro/package.scala (which is not part of this PR, but I've reviewed anyway as part of getting across the project):

  implicit val offsetDateTimeToValue = new ToValue[OffsetDateTime] {
    override def apply(value: OffsetDateTime): String = value.toString
  }

My understanding was that it's not great practice to be using the toString() method for serialization. For long-term compatibility, would it not be better to use format() with an explicit DateTimeFormatter - and then use that same formatter as a second argument to parse(), when deserializing?

oliverlockwood · 2017-08-08T10:51:53Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/config/package.scala

+
+package object config {
+
+  case class AppConfig(scheduler: SchedulerConfig)(implicit system: ActorSystem, materialzer: ActorMaterializer)


Typo: materialzer is missing an i

oliverlockwood · 2017-08-08T14:02:44Z

scheduler/src/main/scala/com/sky/kafka/message/scheduler/SchedulingActor.scala

+    }
+
+    (receiveCreateOrUpdateMessage orElse receiveCancelMessage andThen updateStateAndAck) orElse {
+      case Init => sender ! Ack


The logic priority of the above line isn't 100% clear to me. Is it:

((receiveCreateOrUpdateMessage orElse receiveCancelMessage) andThen updateStateAndAck) orElse

or

(receiveCreateOrUpdateMessage orElse (receiveCancelMessage andThen updateStateAndAck)) orElse

Maybe this is obvious to masters of partial functions, but I don't think that being explicit would hurt?

Also there's an argument for putting the Init case in a named partial function for clarity. I'm less fussed about this though.

this should be clearer now

lacarvalho91 · 2017-08-08T14:41:55Z

scheduler/src/main/scala/com/sky/kafkamessage/scheduler/domain/PublishableMessage.scala

+import com.sky.kafkamessage.scheduler.kafka.ProducerRecordEncoder
+import org.apache.kafka.clients.producer.ProducerRecord
+
+sealed trait PublishableMessage


help me name this pls 😆

SchedulerControlMessage?

with extensions ScheduleDefinition and ScheduleDeletion?

I think ScheduledMessage and ScheduleDeletion are good for the sub-classes just not sure on the name of the trait

lacarvalho91 · 2017-08-08T15:16:31Z

scheduler/src/main/scala/com/sky/kms/SchedulingActor.scala

+
+  case class Cancel(scheduleId: ScheduleId) extends SchedulingMessage
+
+  private case class Trigger(scheduleId: ScheduleId, schedule: Schedule)


i'm not 100% on this name either, maybe its ok

I think it's probably fine

…ference.conf.

lacarvalho91 · 2017-08-08T16:11:12Z

scheduler/src/test/scala/com/sky/kms/AkkaComponentsSpec.scala

+        .settings
+        .supervisionDecider(exception) shouldBe Restart
+
+      verify(metrics).counter(key)


@paoloambrosio-skyuk wasn't sure how we wanted to test these metrics, whether this is ok (using mocks) or starting Kamon and starting an MBeanServer and querying that

…eparately.

oliverlockwood · 2017-08-09T06:26:49Z

README.md

+
+The `schedule-topic` must be configured to use [log compaction](https://kafka.apache.org/documentation/#compaction). 
+This is to allow the scheduler to delete the schedule after it has been written to its destination topic. This is very 
+important because the scheduler uses the `schedule-topic` to reconstruct its state.


It's not only to allow deletion, it's also to ensure that (longer-term) schedules don't get lost after a restart.

oliverlockwood · 2017-08-09T06:27:53Z

build.sbt


  "io.kamon"                   %% "kamon-jmx"                  % kamonVersion,
  "io.kamon"                   %% "kamon-akka-2.5"             % kamonVersion,
  "io.kamon"                   %% "kamon-core"                 % kamonVersion,

  "org.scalatest"              %% "scalatest"                  % "3.0.1"      % Test,
  "com.typesafe.akka"          %% "akka-testkit"               % akkaVersion  % Test,
+  "com.typesafe.akka"          %% "akka-stream-testkit"        % akkaVersion  % Test,
  "net.cakesolutions"          %% "scala-kafka-client-testkit" % kafkaVersion % Test,
  "org.slf4j"                   % "log4j-over-slf4j"           % "1.7.21"     % Test,


As we're a new module I see no reason not to use SLF4J 1.7.25

oliverlockwood · 2017-08-09T06:30:51Z

scheduler/src/main/scala/com/sky/kms/SchedulerApp.scala

+
+  sys.addShutdownHook {
+    logger.info("Kafka Message Scheduler shutting down...")
+    Rewriter.stop(app).value


Restoring @paoloambrosio-skyuk's comment:

All of this is a bit messy Grafter stuff mixed with Akka, Kamon, ... all working in a different way. It would be good to create an abstraction so that we can close everything in the same way.

If we're not addressing that in this PR, perhaps create an issue or at least an issue comment to track, and link to it from here?

oliverlockwood · 2017-08-09T06:37:45Z

scheduler/src/main/scala/com/sky/kms/streams/ScheduledMessagePublisher.scala

+
+/**
+  * Provides a stream that consumes from the queue of triggered messages,
+  * writes the messages to Kafka and then deletes the schedules from Kafka


How about

writes the scheduled messages to the specified Kafka topics, and then deletes the schedules form the scheduling Kafka topic to mark completion

Where were you when I was writing reports at uni!

@lacarvalho91 I was busy giving grammatical advice to developers elsewhere 😉

oliverlockwood · 2017-08-09T07:12:21Z

scheduler/src/main/scala/com/sky/kms/domain/PublishableMessage.scala

+  implicit val scheduleDeletionProducerRecordEnc: ProducerRecordEncoder[ScheduleDeletion] =
+    ProducerRecordEncoder.instance(deletion => new ProducerRecord(deletion.scheduleTopic, deletion.scheduleId.getBytes, null))
+
+  implicit def scheduleDataToProducerRecord(msg: PublishableMessage): ProducerRecord[Array[Byte], Array[Byte]] =


this function naming is a legacy of the old name ScheduleData.

I think it probably needs to be named publishableMessageToProducerRecord.

oliverlockwood · 2017-08-09T07:15:23Z

scheduler/src/main/scala/com/sky/kms/SchedulingActor.scala

+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.FiniteDuration
+
+class SchedulingActor(queue: SourceQueue[(String, ScheduledMessage)], scheduler: Scheduler) extends Actor with ActorLogging {


I actually think that the scheduler parameter should be renamed akkaScheduler.

There's so much "schedule" text dotted around this class that I believe it would increase readability if we clearly identify the Akka scheduling component.

oliverlockwood

I think I'm happy enough with this now 👍

I won't merge yet - I'll give @paoloambrosio-skyuk a chance for a final review first.

lacarvalho91 · 2017-08-09T08:22:24Z

scheduler/src/main/scala/com/sky/kms/SchedulingActor.scala

+        if (cancel(scheduleId, schedules))
+          log.info(s"Cancelled schedule $scheduleId")
+        else
+          log.warning(s"Couldn't cancel $scheduleId")


this log message appears every time a schedule is triggered because we delete the schedule topic after.

What happens is we send the null to the schedule-topic then we process that as a Cancel (to support explicit deletes, and to remove the schedule from the actors state) but the call to cancel fails since the event has already been triggered.

Not sure how we want to handle this, I guess we would want some kind of warning if we fail to cancel something when it should have actually been cancellable

I think this will be non-trivial, so I've raised an issue to cover this: Erroneous WARNING message after every schedule firing

paoloambrosio-skyuk · 2017-08-09T10:16:40Z

scheduler/src/test/scala/com/sky/kms/common/TestActorSystem.scala

@@ -1,4 +1,4 @@
-package common
+package com.sky.kms.common


It might be worth discussing why we put tests in the same package. It has always fascinated me.

Is this not so you can use "package-private" visibilty, and still be able to test it, at least in Java.

Indeed. You should NOT test the internals, and separating packages does that ;-)

robertotena and others added 10 commits August 2, 2017 08:46

Add tests for scheduling messages (WIP)

f397574

scheduling actor complete, just needs to be wired up.

b0d1a69

e2e test.

b115607

Wired up application. TODO: implement publisher stream Flow and reade…

b9a321b

…r stream error handling. Metrics?

Rename scheduler type alias.

26c540e

Scheduling messages and deleting them from the input topic after they…

fe9469d

… have been emitted. Tests passing. TODO: error handling in ScheduleReader stream.

Introduced error handling in ScheduleReader. Just need to provide an …

5f6bbcd

…implicit Sink[ErrorType, _].

Fix splitToScheduleAndMetadata test.

b18e3ec

Logging in publisher stream.

b84b948

Add some clarity around actor receive.

3355089

oliverlockwood reviewed Aug 7, 2017

View reviewed changes

lacarvalho91 changed the title ~~Schedule messages~~ [DON'T MERGE] - Schedule messages Aug 7, 2017

lacarvalho91 and others added 2 commits August 7, 2017 15:37

Fix random data generation error.

1f04e03

Fix memory leak from consecutive test runs.

2958a7b

lacarvalho91 changed the title ~~[DON'T MERGE] - Schedule messages~~ Schedule messages Aug 7, 2017

WIP supervision metrics.

a5e304c

lacarvalho91 commented Aug 7, 2017

View reviewed changes

paoloambrosio-skyuk reviewed Aug 7, 2017

View reviewed changes

paoloambrosio-skyuk reviewed Aug 8, 2017

View reviewed changes

oliverlockwood reviewed Aug 8, 2017

View reviewed changes

paoloambrosio-skyuk reviewed Aug 8, 2017

View reviewed changes

oliverlockwood reviewed Aug 8, 2017

View reviewed changes

Transform schedules to scheduled messages before adding to the queue.

ab143f5

oliverlockwood reviewed Aug 8, 2017

View reviewed changes

Rename package.

33e3e35

lacarvalho91 commented Aug 8, 2017

View reviewed changes

lacarvalho91 force-pushed the schedule-messages branch 3 times, most recently from 8abe011 to 804832a Compare August 8, 2017 15:14

Add custom metrics component

229013a

lacarvalho91 force-pushed the schedule-messages branch from 804832a to 229013a Compare August 8, 2017 15:14

lacarvalho91 commented Aug 8, 2017

View reviewed changes

Remove unused implicits from AppConfig, put some default config in re…

3b12464

…ference.conf.

lacarvalho91 commented Aug 8, 2017

View reviewed changes

Removed monitoring stuff as it has been agreed that it will be done s…

39063b9

…eparately.

lacarvalho91 force-pushed the schedule-messages branch from 572c072 to 96f6125 Compare August 8, 2017 19:48

wait for Init message in actor before handling other messages.

3a81593

lacarvalho91 force-pushed the schedule-messages branch from 96f6125 to 3a81593 Compare August 8, 2017 19:51

oliverlockwood reviewed Aug 9, 2017

View reviewed changes

Address review comments.

837e36c

lacarvalho91 force-pushed the schedule-messages branch from 922bdb0 to 837e36c Compare August 9, 2017 07:53

oliverlockwood approved these changes Aug 9, 2017

View reviewed changes

lacarvalho91 commented Aug 9, 2017

View reviewed changes

oliverlockwood mentioned this pull request Aug 9, 2017

Erroneous WARNING message after every schedule firing #9

Open

paoloambrosio-skyuk reviewed Aug 9, 2017

View reviewed changes

paoloambrosio-skyuk approved these changes Aug 9, 2017

View reviewed changes

paoloambrosio-skyuk merged commit 934d521 into master Aug 9, 2017

paoloambrosio-skyuk deleted the schedule-messages branch August 9, 2017 10:21


		object SchedulerApp extends App with AkkaComponents with LazyLogging {
		object SchedulerApp extends App with LazyLogging with AkkaComponents {


		case class Schedule(time: OffsetDateTime, topic: String, key: Array[Byte], value: Array[Byte]) extends ScheduleData

		case class ScheduleMetadata(scheduleId: ScheduleId, topic: String) extends ScheduleData


		package object config {

		case class AppConfig(scheduler: SchedulerConfig)(implicit system: ActorSystem, materialzer: ActorMaterializer)


		case class Cancel(scheduleId: ScheduleId) extends SchedulingMessage

		private case class Trigger(scheduleId: ScheduleId, schedule: Schedule)

Schedule messages #2

Schedule messages #2

Conversation

lacarvalho91 commented Aug 7, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

paoloambrosio-skyuk left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lacarvalho91 Aug 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lacarvalho91 Aug 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

paoloambrosio-skyuk Aug 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oliverlockwood commented Aug 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oliverlockwood commented Aug 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lacarvalho91 Aug 8, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lacarvalho91 Aug 9, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

oliverlockwood left a comment

Choose a reason for hiding this comment

lacarvalho91 Aug 9, 2017 • edited Loading

Choose a reason for hiding this comment

oliverlockwood Aug 9, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lacarvalho91 commented Aug 7, 2017 •

edited

Loading

lacarvalho91 Aug 8, 2017 •

edited

Loading

lacarvalho91 Aug 8, 2017 •

edited

Loading

paoloambrosio-skyuk Aug 8, 2017 •

edited

Loading

lacarvalho91 Aug 8, 2017 •

edited

Loading

lacarvalho91 Aug 9, 2017 •

edited

Loading

lacarvalho91 Aug 9, 2017 •

edited

Loading

oliverlockwood Aug 9, 2017 •

edited

Loading