Custom aggregator for Spark Structured Streaming from Event Hub

I've created a simple spark structured streaming job which reads events from Azure event hub and aggregates (where the output goes is irrelevant for this post) them per session window. The job is structured like the following snippet (some details have been omitted for readability):

val connStr = new com.microsoft.azure.eventhubs.ConnectionStringBuilder()
           
val customEventhubParameters =
  EventHubsConf(connStr.toString())
  .setMaxEventsPerTrigger(1000)

val incomingStream = spark.readStream.format("eventhubs").options(customEventhubParameters.toMap).load()
  
val eventhubs = incomingStream.select($"body" cast "string")

val testSchema = new StructType()


val messages = incomingStream
  .select($"body".cast(StringType))
  .alias("body")
  .select(from_json($"body",testSchema))
  .select("from_json(body).*")

val windowedCounts = messages
    .withWatermark("timestamp", "10 seconds")
    .groupBy($"body_column", session_window($"timestamp", "1 minute"))
    .count()

Now, I'd like to replace count aggregate function with some custom implementation. I've tried to extend Aggregator (following docs https://spark.apache.org/docs/latest/sql-ref-functions-udf-aggregate.html) and it works for input from a data frame created in the notebook but fails to execute with event hub with exception: Task not serializable: java.io.NotSerializableException: com.microsoft.azure.eventhubs.ConnectionStringBuilder

Is there really no way to create custom aggregators for streaming data?