Remove unchunking evalScan and apply logging to chunks for performance #64
Conversation
| reachedHighest match { | ||
| case Some(highest) => | ||
| HighestOffsetsWithRecord( | ||
| partitionOffsets = t.partitionOffsets - highest, | ||
| consumerRecord = emittableRecord, | ||
| partitionLast = PartitionLast(highest, r.offset).some | ||
| ) |
There was a problem hiding this comment.
If we've reached the highest offset for this partition, store the Topic Partition and Offset inside the HighestOffsetsWithRecord so we can signal this upstream
| .takeWhile(_.partitionOffsets.nonEmpty, takeFailure = true) | ||
| .evalTapChunk(_.partitionLast.traverse { last => | ||
| logger.warn(s"Finished loading data from ${last.topicPartition.show} at offset ${last.offset}") |
There was a problem hiding this comment.
i know this hasn't changed but why is this a warning? this should be info
There was a problem hiding this comment.
Was like that in the Akka lib, always assumed it was intentional 👀 I've lowered it to info
There was a problem hiding this comment.
https://github.com/sky-uk/kafka-topic-loader/blob/0f6c46d858e9d56287eca159eec4a3dc3924ae6c/src/main/scala/uk/sky/kafka/topicloader/TopicLoader.scala#L249 doesn't look like it did, unless i've missed something
There was a problem hiding this comment.
I stand corrected, guess I left it on while debugging and forgot to lower it 😨
| @@ -22,9 +22,12 @@ object TopicLoader extends TopicLoader { | |||
| private[topicloader] given [K, V]: Show[ConsumerRecord[K, V]] = | |||
| Show.show(cr => s"${cr.topic}-${cr.partition}") | |||
|
|
|||
| private case class PartitionLast(topicPartition: TopicPartition, offset: Long) | |||
There was a problem hiding this comment.
PartitionLastOffset might make this clearer?
bcarter97
force-pushed
the
60-remove-unchunking-scan
branch
from
ad4363f to
66541f3
Compare
Description
Currently we use
evalScanto perform the logic of producing a ConsumerRecord and the non-empty partition map. this is effectful as we perform logging upon reaching the highest offset.It would be more efficient to use the non-effectful scan, as
evalScande-chunks a stream into singleton chunks. There unfortunately is noevalScanChunksequivalent ofscanChunks.We still perform the logging, however we can operate on the chunk for performance by using
evalTapChunkRelated Issues
Definition of Done:
The Below tasks should be completed before marking a PR as ready for review.