Mirrors/the-algorithm
2
0
Fork 0
mirror of https://github.com/twitter/the-algorithm.git synced 2025-01-02 23:51:53 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge 832a5de758 into 72eda9a24f

Browse Source
This commit is contained in:
Rajiv Singh 2023-07-17 21:41:35 -05:00 committed by GitHub
commit 9bc99f431f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/scala/com/twitter/recos/hose/common/UnifiedGraphWriter.scala
View File

@ -85,16 +85,13 @@ trait UnifiedGraphWriter[
}
}
private def initRecosHoseKafka(
queue: java.util.Queue[Array[RecosHoseMessage]],
queuelimit: Semaphore,
): Unit = {
private def initRecosHoseKafka(queue: java.util.Queue[Array[RecosHoseMessage]], queuelimit: Semaphore): Unit = {
try {
consumers = (0 until consumerNum).map { index =>
new ThreadSafeKafkaConsumerClient(
kafkaConsumerBuilder.clientId(s"clientId-$index").enableAutoCommit(false).config)
val consumers = (0 until consumerNum).map { index =>
new ThreadSafeKafkaConsumerClient(kafkaConsumerBuilder.clientId(s"clientId-$index").enableAutoCommit(false).config)
}
processors = consumers.zipWithIndex.map {
val processors = consumers.zipWithIndex.map {
case (consumer, index) =>
val bufferedWriter = BufferedEdgeCollector(bufferSize, queue, queuelimit, statsReceiver)
val processor = RecosEdgeProcessor(bufferedWriter)(statsReceiver)
@ -112,21 +109,14 @@ trait UnifiedGraphWriter[
}
log.info(s"starting ${processors.size} recosKafka processors")
processors.foreach { processor =>
processor.start()
}
processors.foreach(_.start())
} catch {
case e: Throwable =>
e.printStackTrace()
log.error(e, e.toString)
processors.foreach { processor =>
processor.close()
}
processors = Seq.empty
consumers.foreach { consumer =>
consumer.close()
}
consumers = Seq.empty
processors.foreach(_.close())
consumers.foreach(_.close())
}
}