14645650020

Committed 24 Apr 2025 03:34PM CUT coverage: 94.106% (+0.08%) from 94.024%

Build Type

Committed by

Commit Message

Merge 5b4eadf6e into 350691e40

Pull Request Pull Request #760: Update pureconfig-cats, pureconfig-core, ... to 0.17.9

Run Details

5269 of 5599 relevant lines covered (94.11%)

1.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.7

/replicator/src/main/scala/com/evolutiongaming/kafka/journal/replicator/TopicReplicator.scala

package com.evolutiongaming.kafka.journal.replicator

import cats.Monad
import cats.data.{NonEmptyList as Nel, NonEmptyMap as Nem, NonEmptySet as Nes}
import cats.effect.*
import cats.effect.implicits.*
import cats.implicits.*
import com.evolutiongaming.catshelper.*
import com.evolutiongaming.catshelper.ClockHelper.*
import com.evolutiongaming.catshelper.ParallelHelper.*
import com.evolutiongaming.kafka.journal.*
import com.evolutiongaming.kafka.journal.conversions.{ConsRecordToActionRecord, KafkaRead}
import com.evolutiongaming.kafka.journal.eventual.*
import com.evolutiongaming.kafka.journal.util.Fail
import com.evolutiongaming.kafka.journal.util.SkafkaHelper.*
import com.evolutiongaming.retry.Sleep
import com.evolutiongaming.skafka.*
import com.evolutiongaming.skafka.consumer.{AutoOffsetReset, ConsumerConfig}
import scodec.bits.ByteVector

import java.time.Instant
import scala.concurrent.duration.*
import scala.util.Try

/**
 * Consumes the Kafka topic and "splits" the data stream into [[PartitionFlow]]s and "splits" each
 * per-partition stream in [[KeyFlow]]s.
 *
 * Basically:
 *   - result of each Kafka's `poll` gets grouped per partition and key
 *   - grouped per-key records are processed by [[ReplicateRecords]]
 */
private[journal] object TopicReplicator {

  def make[F[_]: Concurrent: Sleep: ToTry: LogOf: Fail: MeasureDuration: JsonCodec](
    topic: Topic,
    journal: ReplicatedJournal[F],
    consumer: Resource[F, TopicConsumer[F]],
    metrics: TopicReplicatorMetrics[F],
    cacheOf: CacheOf[F],
    replicatedOffsetNotifier: ReplicatedOffsetNotifier[F],
  ): Resource[F, F[Outcome[F, Throwable, Unit]]] = {

    implicit val fromAttempt: FromAttempt[F] = FromAttempt.lift[F]
    implicit val fromJsResult: FromJsResult[F] = FromJsResult.lift[F]
    implicit val jsonCodec: JsonCodec[Try] = JsonCodec.summon[F].mapK(ToTry.functionK)

    val kafkaRead = KafkaRead.summon[F, Payload]
    val eventualWrite = EventualWrite.summon[F, Payload]

    def consume(
      consumer: Resource[F, TopicConsumer[F]],
      log: Log[F],
    ) = {

      val consRecordToActionRecord = ConsRecordToActionRecord[F]
      of(
        topic = topic,
        consumer = consumer,
        consRecordToActionRecord = consRecordToActionRecord,
        kafkaRead = kafkaRead,
        eventualWrite = eventualWrite,
        journal = journal,
        metrics = metrics,
        log = log,
        cacheOf = cacheOf,
        replicatedOffsetNotifier = replicatedOffsetNotifier,
      )
    }

    for {
      log <- topicLoggerOf(topic).toResource
      done <- consume(consumer, log).background
    } yield done
  }

  private def topicLoggerOf[F[_]: LogOf: Monad](topic: Topic): F[Log[F]] = {
    LogOf[F].apply(TopicReplicator.getClass).map(_ prefixed topic)
  }

  def of[F[_]: Concurrent: Sleep: MeasureDuration, A](
    topic: Topic,
    consumer: Resource[F, TopicConsumer[F]],
    consRecordToActionRecord: ConsRecordToActionRecord[F],
    kafkaRead: KafkaRead[F, A],
    eventualWrite: EventualWrite[F, A],
    journal: ReplicatedJournal[F],
    metrics: TopicReplicatorMetrics[F],
    log: Log[F],
    cacheOf: CacheOf[F],
    replicatedOffsetNotifier: ReplicatedOffsetNotifier[F],
  ): F[Unit] = {

    trait PartitionFlow {
      def apply(timestamp: Instant, records: Nel[ConsRecord]): F[Unit]
    }

    trait KeyFlow {
      def apply(timestamp: Instant, records: Nel[ConsRecord]): F[Int]
    }

    val topicFlowOf: TopicFlowOf[F] = { (topic: Topic) =>
      {
        for {
          journal <- journal.journal(topic)
          cache <- cacheOf[Partition, PartitionFlow](topic)
        } yield {

          def remove(partitions: Nes[Partition]) = {
            partitions.parFoldMap1 { partition => cache.remove(partition) }
          }

          new TopicFlow[F] {

            def assign(partitions: Nes[Partition]) = {
              log.info(s"assign ${ partitions.mkString_(",") }")
            }

            def apply(records: Nem[Partition, Nel[ConsRecord]]) = {
              for {
                duration <- MeasureDuration[F].start
                timestamp <- Clock[F].instant
                _ <- records.parFoldMap1 {
                  case (partition, records) =>
                    for {
                      partitionFlow <- cache.getOrUpdate(partition) {
                        for {
                          journal <- journal(partition)
                          offsets = journal.offsets
                          offsetRef <- Resource.eval {
                            for {
                              offset <- offsets.get
                              ref <- Ref.of(offset)
                            } yield ref
                          }
                          cache <- cacheOf[String, KeyFlow](topic)
                        } yield { (timestamp: Instant, records: Nel[ConsRecord]) =>
                          {
                            for {
                              offset <- offsetRef.get
                              _ <- offset
                                .fold {
                                  records.some
                                } { offset =>
                                  records.filter { _.offset > offset }.toNel
                                }
                                .foldMapM { records =>
                                  records
                                    .groupBy { _.key.map { _.value } }
                                    .parFoldMap1 {
                                      case (key, records) =>
                                        key.foldMapM { key =>
                                          for {
                                            keyFlow <- cache.getOrUpdate(key) {
                                              journal
                                                .journal(key)
                                                .map { journal =>
                                                  val replicateRecords = ReplicateRecords(
                                                    consRecordToActionRecord,
                                                    journal,
                                                    metrics,
                                                    kafkaRead,
                                                    eventualWrite,
                                                    log,
                                                  )
                                                  (timestamp: Instant, records: Nel[ConsRecord]) => {
                                                    replicateRecords(records, timestamp)
                                                  }
                                                }
                                            }
                                            result <- keyFlow(timestamp, records)
                                          } yield result
                                        }
                                    }
                                }
                              result <- {
                                val offset1 = records.maximumBy { _.offset }.offset

                                def setAndNotify = offsetRef.set(offset1.some) >>
                                  replicatedOffsetNotifier.onReplicatedOffset(TopicPartition(topic, partition), offset1)

                                offset.fold {
                                  for {
                                    a <- offsets.create(offset1, timestamp)
                                    _ <- setAndNotify
                                  } yield a
                                } { offset =>
                                  if (offset1 > offset) {
                                    for {
                                      a <- offsets.update(offset1, timestamp)
                                      _ <- setAndNotify
                                    } yield a
                                  } else {
                                    ().pure[F]
                                  }
                                }
                              }
                            } yield result
                          }
                        }
                      }
                      result <- partitionFlow(timestamp, records)
                    } yield result
                }
                duration <- duration
                _ <- metrics.round(duration, records.foldLeft(0) { _ + _.size })
              } yield {
                records.map { records =>
                  records.foldLeft { Offset.min } { (offset, record) =>
                    record
                      .offset
                      .inc[Try]
                      .fold(_ => offset, _ max offset)
                  }
                }.toSortedMap
              }
            }

            def revoke(partitions: Nes[Partition]) = {
              for {
                _ <- log.info(s"revoke ${ partitions.mkString_(",") }")
                a <- remove(partitions)
              } yield a
            }

            def lose(partitions: Nes[Partition]) = {
              for {
                _ <- log.info(s"lose ${ partitions.mkString_(",") }")
                a <- remove(partitions)
              } yield a
            }
          }
        }
      }
    }

    ConsumeTopic(topic, consumer, topicFlowOf, log)
  }

  object ConsumerOf {

    def make[F[_]: Temporal: KafkaConsumerOf: FromTry: LogOf](
      topic: Topic,
      config: ConsumerConfig,
      pollTimeout: FiniteDuration,
      hostName: Option[HostName],
    ): Resource[F, TopicConsumer[F]] = {

      val groupId = {
        val prefix = config.groupId getOrElse "replicator"
        s"$prefix-$topic"
      }

      val common = config.common

      val clientId = {
        val clientId = common.clientId getOrElse "replicator"
        hostName.fold(clientId) { hostName => s"$clientId-$hostName" }
      }

      val config1 = config.copy(
        common = common.copy(clientId = clientId.some),
        groupId = groupId.some,
        autoOffsetReset = AutoOffsetReset.Earliest,
        autoCommit = false,
      )

      for {
        consumer <- KafkaConsumerOf[F].apply[String, ByteVector](config1)
        log <- topicLoggerOf[F](topic).toResource
        metadata = hostName.fold { Metadata.empty } { _.value }
        commit <- TopicCommit.asyncPeriodic(
          topic = topic,
          commitMetadata = metadata,
          commitPeriod = 5.seconds,
          consumer = consumer,
          log = log,
        )
      } yield {
        TopicConsumer(topic, pollTimeout, commit, consumer)
      }
    }
  }
}

1	package com.evolutiongaming.kafka.journal.replicator
2
3	import cats.Monad
4	import cats.data.{NonEmptyList as Nel, NonEmptyMap as Nem, NonEmptySet as Nes}
5	import cats.effect.*
6	import cats.effect.implicits.*
7	import cats.implicits.*
8	import com.evolutiongaming.catshelper.*
9	import com.evolutiongaming.catshelper.ClockHelper.*
10	import com.evolutiongaming.catshelper.ParallelHelper.*
11	import com.evolutiongaming.kafka.journal.*
12	import com.evolutiongaming.kafka.journal.conversions.{ConsRecordToActionRecord, KafkaRead}
13	import com.evolutiongaming.kafka.journal.eventual.*
14	import com.evolutiongaming.kafka.journal.util.Fail
15	import com.evolutiongaming.kafka.journal.util.SkafkaHelper.*
16	import com.evolutiongaming.retry.Sleep
17	import com.evolutiongaming.skafka.*
18	import com.evolutiongaming.skafka.consumer.{AutoOffsetReset, ConsumerConfig}
19	import scodec.bits.ByteVector
20
21	import java.time.Instant
22	import scala.concurrent.duration.*
23	import scala.util.Try
24
25	/**
26	* Consumes the Kafka topic and "splits" the data stream into [[PartitionFlow]]s and "splits" each
27	* per-partition stream in [[KeyFlow]]s.
28	*
29	* Basically:
30	* - result of each Kafka's `poll` gets grouped per partition and key	1✔
31	* - grouped per-key records are processed by [[ReplicateRecords]]
32	*/
33	private[journal] object TopicReplicator {
34
35	def make[F[_]: Concurrent: Sleep: ToTry: LogOf: Fail: MeasureDuration: JsonCodec](	1✔
36	topic: Topic,	1✔
37	journal: ReplicatedJournal[F],	1✔
38	consumer: Resource[F, TopicConsumer[F]],
39	metrics: TopicReplicatorMetrics[F],
40	cacheOf: CacheOf[F],
41	replicatedOffsetNotifier: ReplicatedOffsetNotifier[F],
42	): Resource[F, F[Outcome[F, Throwable, Unit]]] = {	1✔
43
44	implicit val fromAttempt: FromAttempt[F] = FromAttempt.lift[F]	2✔
45	implicit val fromJsResult: FromJsResult[F] = FromJsResult.lift[F]	2✔
46	implicit val jsonCodec: JsonCodec[Try] = JsonCodec.summon[F].mapK(ToTry.functionK)	2✔
47
48	val kafkaRead = KafkaRead.summon[F, Payload]	2✔
49	val eventualWrite = EventualWrite.summon[F, Payload]	2✔
50		1✔
51	def consume(	1✔
52	consumer: Resource[F, TopicConsumer[F]],
53	log: Log[F],	1✔
54	) = {	1✔
55		1✔
56	val consRecordToActionRecord = ConsRecordToActionRecord[F]	1✔
57	of(	2✔
58	topic = topic,	1✔
59	consumer = consumer,
60	consRecordToActionRecord = consRecordToActionRecord,
61	kafkaRead = kafkaRead,
62	eventualWrite = eventualWrite,	1✔
63	journal = journal,
64	metrics = metrics,
65	log = log,
66	cacheOf = cacheOf,
67	replicatedOffsetNotifier = replicatedOffsetNotifier,
68	)
69	}
70
71	for {	1✔
72	log <- topicLoggerOf(topic).toResource	1✔
73	done <- consume(consumer, log).background	2✔
74	} yield done
75	}
76
77	private def topicLoggerOf[F[_]: LogOf: Monad](topic: Topic): F[Log[F]] = {	1✔
78	LogOf[F].apply(TopicReplicator.getClass).map(_ prefixed topic)	2✔
79	}
80
81	def of[F[_]: Concurrent: Sleep: MeasureDuration, A](
82	topic: Topic,
83	consumer: Resource[F, TopicConsumer[F]],
84	consRecordToActionRecord: ConsRecordToActionRecord[F],
85	kafkaRead: KafkaRead[F, A],
86	eventualWrite: EventualWrite[F, A],
87	journal: ReplicatedJournal[F],
88	metrics: TopicReplicatorMetrics[F],
89	log: Log[F],
90	cacheOf: CacheOf[F],
91	replicatedOffsetNotifier: ReplicatedOffsetNotifier[F],
92	): F[Unit] = {
93
94	trait PartitionFlow {
95	def apply(timestamp: Instant, records: Nel[ConsRecord]): F[Unit]
96	}
97
98	trait KeyFlow {
99	def apply(timestamp: Instant, records: Nel[ConsRecord]): F[Int]
100	}
101
102	val topicFlowOf: TopicFlowOf[F] = { (topic: Topic) =>
103	{
104	for {	1✔
105	journal <- journal.journal(topic)	2✔
106	cache <- cacheOf[Partition, PartitionFlow](topic)	2✔
107	} yield {
108
109	def remove(partitions: Nes[Partition]) = {	1✔
110	partitions.parFoldMap1 { partition => cache.remove(partition) }	2✔
111	}
112
113	new TopicFlow[F] {	1✔
114
115	def assign(partitions: Nes[Partition]) = {
116	log.info(s"assign ${ partitions.mkString_(",") }")	1✔
117	}
118
119	def apply(records: Nem[Partition, Nel[ConsRecord]]) = {	1✔
120	for {	1✔
121	duration <- MeasureDuration[F].start	2✔
122	timestamp <- Clock[F].instant	2✔
123	_ <- records.parFoldMap1 {	2✔
124	case (partition, records) =>	1✔
125	for {	1✔
126	partitionFlow <- cache.getOrUpdate(partition) {	2✔
127	for {	1✔
128	journal <- journal(partition)	2✔
129	offsets = journal.offsets	2✔
130	offsetRef <- Resource.eval {	2✔
131	for {	1✔
132	offset <- offsets.get	2✔
133	ref <- Ref.of(offset)	2✔
134	} yield ref
135	}
136	cache <- cacheOf[String, KeyFlow](topic)	2✔
137	} yield { (timestamp: Instant, records: Nel[ConsRecord]) =>
138	{
139	for {	1✔
140	offset <- offsetRef.get	2✔
141	_ <- offset	2✔
142	.fold {
143	records.some	2✔
144	} { offset =>	2✔
145	records.filter { _.offset > offset }.toNel	2✔
146	}
147	.foldMapM { records =>	1✔
148	records	1✔
149	.groupBy { _.key.map { _.value } }	2✔
150	.parFoldMap1 {	1✔
151	case (key, records) =>	1✔
152	key.foldMapM { key =>	2✔
153	for {	1✔
154	keyFlow <- cache.getOrUpdate(key) {	2✔
155	journal	1✔
156	.journal(key)	1✔
157	.map { journal =>	2✔
158	val replicateRecords = ReplicateRecords(	2✔
159	consRecordToActionRecord,	1✔
160	journal,	1✔
161	metrics,	1✔
162	kafkaRead,	1✔
163	eventualWrite,	1✔
164	log,	1✔
165	)	1✔
166	(timestamp: Instant, records: Nel[ConsRecord]) => {
167	replicateRecords(records, timestamp)	2✔
168	}
169	}	1✔
170	}	1✔
171	result <- keyFlow(timestamp, records)	2✔
172	} yield result	1✔
173	}	1✔
174	}	1✔
175	}	1✔
176	result <- {	2✔
177	val offset1 = records.maximumBy { _.offset }.offset	2✔
178		1✔
179	def setAndNotify = offsetRef.set(offset1.some) >>	1✔
180	replicatedOffsetNotifier.onReplicatedOffset(TopicPartition(topic, partition), offset1)	1✔
181
182	offset.fold {	1✔
183	for {	1✔
184	a <- offsets.create(offset1, timestamp)	2✔
185	_ <- setAndNotify	2✔
186	} yield a	2✔
187	} { offset =>	2✔
188	if (offset1 > offset) {	2✔
189	for {	1✔
190	a <- offsets.update(offset1, timestamp)	2✔
191	_ <- setAndNotify	2✔
192	} yield a	2✔
193	} else {	1✔
194	().pure[F]	2✔
195	}
196	}
197	}
198	} yield result
199	}
200	}	1✔
201	}	1✔
202	result <- partitionFlow(timestamp, records)	2✔
203	} yield result	1✔
204	}	1✔
205	duration <- duration	2✔
206	_ <- metrics.round(duration, records.foldLeft(0) { _ + _.size })	2✔
207	} yield {	1✔
208	records.map { records =>	2✔
209	records.foldLeft { Offset.min } { (offset, record) =>	2✔
210	record	1✔
211	.offset	2✔
212	.inc[Try]	2✔
213	.fold(_ => offset, _ max offset)	1✔
214	}
215	}.toSortedMap	2✔
216	}	1✔
217	}	1✔
218
219	def revoke(partitions: Nes[Partition]) = {	1✔
220	for {	1✔
221	_ <- log.info(s"revoke ${ partitions.mkString_(",") }")	2✔
222	a <- remove(partitions)	2✔
223	} yield a
224	}	1✔
225		1✔
226	def lose(partitions: Nes[Partition]) = {	1✔
227	for {	1✔
228	_ <- log.info(s"lose ${ partitions.mkString_(",") }")	×
229	a <- remove(partitions)	×
230	} yield a	1✔
231	}	1✔
232	}
233	}	1✔
234	}	1✔
235	}	1✔
236
237	ConsumeTopic(topic, consumer, topicFlowOf, log)	2✔
238	}
239
240	object ConsumerOf {
241
242	def make[F[_]: Temporal: KafkaConsumerOf: FromTry: LogOf](	1✔
243	topic: Topic,
244	config: ConsumerConfig,
245	pollTimeout: FiniteDuration,	1✔
246	hostName: Option[HostName],	1✔
247	): Resource[F, TopicConsumer[F]] = {	1✔
248
249	val groupId = {
250	val prefix = config.groupId getOrElse "replicator"	2✔
251	s"$prefix-$topic"	1✔
252	}	1✔
253
254	val common = config.common	1✔
255
256	val clientId = {
257	val clientId = common.clientId getOrElse "replicator"	2✔
258	hostName.fold(clientId) { hostName => s"$clientId-$hostName" }	2✔
259	}
260
261	val config1 = config.copy(	1✔
262	common = common.copy(clientId = clientId.some),	2✔
263	groupId = groupId.some,	2✔
264	autoOffsetReset = AutoOffsetReset.Earliest,	1✔
265	autoCommit = false,	1✔
266	)
267
268	for {	1✔
269	consumer <- KafkaConsumerOf[F].apply[String, ByteVector](config1)	2✔
270	log <- topicLoggerOf[F](topic).toResource	2✔
271	metadata = hostName.fold { Metadata.empty } { _.value }	2✔
272	commit <- TopicCommit.asyncPeriodic(	2✔
273	topic = topic,
274	commitMetadata = metadata,
275	commitPeriod = 5.seconds,	2✔
276	consumer = consumer,
277	log = log,
278	)
279	} yield {
280	TopicConsumer(topic, pollTimeout, commit, consumer)	2✔
281	}
282	}
283	}
284	}	1✔

evolution-gaming / kafka-journal / 14645650020

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous