fix test

evolution-gaming · Nov 2, 2023 · e19f7e6 · e19f7e6
1 parent e074b57
commit e19f7e6
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 7 deletions.
diff --git a/journal/src/main/scala/com/evolutiongaming/kafka/journal/HeadCache.scala b/journal/src/main/scala/com/evolutiongaming/kafka/journal/HeadCache.scala
@@ -16,6 +16,7 @@ import com.evolutiongaming.smetrics.MetricsHelper._
 import com.evolutiongaming.smetrics._
 
 import java.nio.ByteBuffer
+import java.util.UUID
 import scala.concurrent.duration._
 
 /** Metainfo of events written to Kafka, but not yet replicated to Cassandra.
@@ -132,7 +133,7 @@ object HeadCache {
         metrics,
         pointer = KafkaConsumerOf[F].apply[Partition, Offset](
           consumerConfig.copy(
-            groupId            = none,
+            groupId            = UUID.randomUUID().toString.some,
             autoCommit         = true,
             autoCommitInterval = 5.seconds.some,
             autoOffsetReset    = AutoOffsetReset.Latest,

diff --git a/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadCacheSpec.scala b/journal/src/test/scala/com/evolutiongaming/kafka/journal/HeadCacheSpec.scala
@@ -16,7 +16,7 @@ import com.evolutiongaming.smetrics.CollectorRegistry
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AsyncWordSpec
 import com.evolutiongaming.retry.Sleep
-import com.evolutiongaming.skafka.consumer.{ConsumerRecord, ConsumerRecords}
+import com.evolutiongaming.skafka.consumer.{Consumer, ConsumerRecord, ConsumerRecords}
 
 import scala.collection.immutable.Queue
 import scala.concurrent.duration._
@@ -188,7 +188,7 @@ class HeadCacheSpec extends AsyncWordSpec with Matchers {
     "timeout" in {
       val consumer   = TopicCache.Consumer.empty[IO]
       val headCache  = headCacheOf(
-        HeadCache.Eventual.empty, 
+        HeadCache.Eventual.empty,
         consumer.pure[IO].toResource,
         config.copy(timeout = 10.millis))
       val result = headCache.use { headCache =>
@@ -276,7 +276,7 @@ object HeadCacheSpec {
       headers = headers).get
   }
 
-  
+
   implicit val LogIO: Log[IO] = Log.empty[IO]
 
 
@@ -294,7 +294,7 @@ object HeadCacheSpec {
         eventual = eventual,
         consumer = consumer,
         metrics = metrics.some,
-        pointer = ???
+        pointer = KafkaConsumer.of(Consumer.empty[IO, Partition, Offset].pure[Resource[IO, *]])
       )
     } yield headCache
   }
@@ -377,7 +377,7 @@ object HeadCacheSpec {
       }
     }
   }
-  
-  
+
+
   case object TestError extends RuntimeException with NoStackTrace
 }