Kafka source recover from RebalanceInProgressException

When a consumer group rebalances, the Kafka consumer must commit offsets
before partitions are revoked.

Previously we were using "Eager" rebalance revoke mode from fs2-kafka,
with the following consequences:

1. Partitions were revoked immediately without waiting for the
   application to commit offsets. This could lead to duplicates.
2. The commit occasionally failed with a RebalanceInProgressException,
   if the application tried to commit offsets after the rebalance had
   started.

This change switches to "Graceful" rebalance revoke mode. It works as
follows:

1. Source waits up to session.timeout.ms for the fs2 stream to
   finalize. This includes committing outstanding offsets.
2. Rebalancing only proceeds after the fs2 stream has finalized, or
   after session.timeout.ms. This reduces the possibility of duplicates
   due to re-processing.
3. We catch and ignore the RebalanceInProgressException in case the
   downstream application cannot finalize the fs2 stream within the
   session timeout. This is needed e.g. for Lake Loader which is slow
   to finalize a window.

Author: istreeter

modules/kafka/src/main/scala/com/snowplowanalytics/snowplow/streams/kafka/source/KafkaSource.scala

@@ -22,6 +22,7 @@ import scala.concurrent.duration.{DurationLong, FiniteDuration}
 import fs2.kafka._
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.clients.consumer.OffsetAndMetadata
+import org.apache.kafka.common.errors.RebalanceInProgressException
 
 // snowplow
 import com.snowplowanalytics.snowplow.streams.SourceAndAck
@@ -68,7 +69,9 @@ private[kafka] object KafkaSource {
 
       val empty: KafkaCheckpoints = Map.empty
       def ack(c: KafkaCheckpoints): F[Unit] =
-        kafkaConsumer.commitSync(c)
+        kafkaConsumer.commitSync(c).recoverWith { case _: RebalanceInProgressException =>
+          Logger[F].warn("Failed to commit offsets during rebalance, offsets will be lost and events may be reprocessed")
+        }
       def nack(c: KafkaCheckpoints): F[Unit] = Applicative[F].unit
     }
 
@@ -145,4 +148,5 @@ private[kafka] object KafkaSource {
       .withProperties(config.consumerConf)
       .withEnableAutoCommit(false)
       .withCommitTimeout(config.commitTimeout)
+      .withRebalanceRevokeMode(RebalanceRevokeMode.Graceful)
 }