laserdisc-io / tamer Goto Github PK

There might be an edge case when there is only one message and the program is stuck waiting for other states at startup before proceeding (in the plainStream instruction). I will try to reproduce this later, this is just a note.

Tamer tends to be used concurrently and in swarm in a cluster. The logs don't include contextual information about who is doing what. Starting from

@barambani noticed with S3 module tamer uses 100% cpu.
I suspect because of

#64 (14m871s871268ms) 
   Status: Running()
Fiber:Id(1613749779114,64) was supposed to continue to:
  a future continuation at zio.ZIO.forever(ZIO.scala:686)

Fiber:Id(1613749779114,64) execution trace:
  at zio.ZIO.forever(ZIO.scala:686)
  at zio.ZIO.unit(ZIO.scala:2014)
  at log.effect.LogWriterConstructor$$anon$1$$anon$2.write(LogWriterConstructor.scala:31)
  at tamer.kafka.Kafka$$anon$1.runLoop(Kafka.scala:66)
  at zio.ZQueue$$anon$4.takeAll(ZQueue.scala:627)
  at zio.ZIO$.effectSuspendTotal(ZIO.scala:2662)
  ...

Do you think this is the problem?
Should we solve it by using take, which suspends, instead of takeAll that continuously returning an empty list?

Hook in publish to schema registry on write

Since tpolecat/doobie@96594de

Really tamer should have test coverage to catch these... 🙄

While connection timeout in STTP is a global setting, and thus easily configurable, read timeout is per request, this means there is no way to set it using the simplified API (because the query builder is provided).
It would be cool to add it as an additional optional setting.

Tamer is designed to always go as fast as possible. Sometimes it may be desirable to slow down, e.g. once we caught up with the "initial load".. We could use a scheduler controlled in the Setup and returned by it so to slow this down, when needed

Having a TamerError here is wrong because it leaks in user code.

Will do a PR for this soon.

I noticed there are no S3 tags and the descriptions says "Standalone alternative to Kafka JDBC Connector"
I propose to update it to "Standalone alternative to Kafka JDBC Connector and S3 Kafka Connector"

The reason is nanoTime gives you a monotonic value (representing the number of nanoseconds since system boot) while instant gives you a non-monotonic value (millis since epoch). The latter is subject to changes made to the system clock (including it going backwards), usually as result of NTP adjustments. See also https://medium.com/@kralka/its-about-time-d42e68dbe1df

Originally posted by @sirocchj in #212 (comment)

Here TamerError leaks to the user, a quick and dirty solution would be to widen to a RIO

https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/#define-a-liveness-http-request

After #540 there are no more checks of the state topic in case of crashes. On one hand the cases are greatly reduced by the introduction of transactionality, on the other this does nothing to prevent external factors (wrong number of replicas or simply external producers to the topic).
We could add an invariant check, that not only checks the topic for consistency on start up, but also continuously monitors the number of messages read and stops in case more than one message is in the buffer.

val authentication: Authentication[Any] = new Authentication[Any] {
    override def addAuthentication(request: SttpRequest, supplementalSecret: Option[String]): SttpRequest =
    // ...
}

A suggestion is to make the type parameter -R <: SttpClient or to make authentication independent from any particular client.

It currently could be updated before the last batch (in queue) has been published to Kafka. Likely solvable with a semaphore

The field below, while logically belongs to core, it's never used by it and only used by the modules. This is just code smell and not a real problem but might indicate there is a better way to structure the code.

override val recordKey: (State, LastAlertTime) => LastAlertTimeKey = ???

and make the code still work because you are using a different way to create your record key.