I've run into an issue with finite fs2-kafka streams and zipping. It seems that if I take more than 1000 elements from the default-configured streams and then zip them, the first stream's elements get recycled after the 1000th one starting with the 0th one. The example code snippet makes it clearer below.
Scala version: 2.12.18
fs2.kafka version: 3.1.0
{
val bootstrapServers = kafkaHostPort
val topic1Name = "test-topic-1"
val topic2Name = "test-topic-2"
val producerSettings: ProducerSettings[IO, String, String] =
ProducerSettings(
keySerializer = Serializer.string[IO](StandardCharsets.UTF_8),
valueSerializer = Serializer.string[IO](StandardCharsets.UTF_8),
)
.withBootstrapServers(bootstrapServers)
.withEnableIdempotence(true)
.withRetries(3)
val consumerSettings: ConsumerSettings[IO, String, String] =
ConsumerSettings(
Deserializer.string[IO],
Deserializer.string[IO],
)
.withAutoOffsetReset(AutoOffsetReset.Earliest)
.withBootstrapServers(bootstrapServers)
.withGroupId("test-group")
val records = (0 to 1010).flatMap { i =>
Vector(
ProducerRecord(topic1Name, s"k${i}", s"v${i}"),
ProducerRecord(topic2Name, s"k${i}", s"v${i}"),
)
}.toVector
val producerRecords = ProducerRecords(records)
KafkaProducer.resource(producerSettings).use { producer =>
for {
_ <- producer.produce(producerRecords)
topic1Stream = KafkaConsumer
.stream(consumerSettings)
.subscribeTo(topic1Name)
.records
.take(1005)
topic2Stream = KafkaConsumer
.stream(consumerSettings)
.subscribeTo(topic2Name)
.records
.take(1005)
result <- (topic1Stream zip topic2Stream)
.evalTap { case (lhs, rhs) => IO.println(lhs) >> IO.println(rhs) }
.compile
.toVector
} yield {
result.foreach { case (lhs, rhs) =>
assert(lhs.record.key == rhs.record.key && lhs.record.value == rhs.record.value)
}
}
}
}
This code fails on the assert with "k[]0" did not equal "k[100]0". I added some logging to see more clearly what is happening, and there it shows this:
Hello,
I've run into an issue with finite fs2-kafka streams and zipping. It seems that if I take more than 1000 elements from the default-configured streams and then zip them, the first stream's elements get recycled after the 1000th one starting with the 0th one. The example code snippet makes it clearer below.
Scala version: 2.12.18 fs2.kafka version: 3.1.0
This code fails on the assert with
"k[]0" did not equal "k[100]0"
. I added some logging to see more clearly what is happening, and there it shows this:Could you help me resolve this?
P.S.: this other issue might be related: https://github.com/fd4s/fs2-kafka/issues/1292