From d4596cd6e9cc0ae99aef2c6911714edb16ee723f Mon Sep 17 00:00:00 2001 From: Brian Harrington Date: Thu, 8 Feb 2024 10:22:49 -0600 Subject: [PATCH] lwcapi: batch sub messages When setting up a new subscription, batch the subscription metadata messages sent to the queue. Otherwise if there is a large set it they could get written quickly enough to exceed the queue size. --- .../main/scala/com/netflix/atlas/lwcapi/SubscribeApi.scala | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/SubscribeApi.scala b/atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/SubscribeApi.scala index 349fde1b6..2d2f92457 100644 --- a/atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/SubscribeApi.scala +++ b/atlas-lwcapi/src/main/scala/com/netflix/atlas/lwcapi/SubscribeApi.scala @@ -195,11 +195,12 @@ class SubscribeApi( // Add any new expressions val (queue, addedSubs) = sm.subscribe(streamId, splits) - addedSubs.foreach { sub => + val subMessages = addedSubs.map { sub => val meta = sub.metadata val exprInfo = LwcDataExpr(meta.id, meta.expression, meta.frequency) - queue.offer(Seq(LwcSubscription(expr.expression, List(exprInfo)))) + LwcSubscription(expr.expression, List(exprInfo)) } + queue.offer(subMessages) // Add expression ids in use by this split subIdsBuilder ++= splits.map(_.metadata.id)