有 akka 大佬吗， 请教一个问题

用 slick 流式读取 mysql 一个大表 700w, 在 map 的时候计数，发现到 360w 就停了， 请问下这是什么原因?

val source: Source[Event, NotUsed] = Slick.source(query1)

val sink: Sink[ProducerRecord[String, Event], Future[Done]] = Producer.plainSink(p)

var counter = 0;

val done: Future[Done] =
        source
          .map(
            e =>
              {
                counter += 1
                println(counter)
                new ProducerRecord(topic, e.sid, EventFactory.build_event(e))
              }
).runWith(sink)