Getting unspecified value parameter scala Kafka

Question

Hi I am getting this compilation error -

Unspecified Value parameters: aggregator: (String, String, NotInferedVR) => NotInferedVR

But I clearly have an aggregator there already. Does anyone know what's going on?

 val stream = builder
    .stream(inputTopic)(Consumed.`with`(Serdes.String(), Serdes.ByteArray()))
    .map{ (key: String, value: Array[Byte]) =>
      println(s"key = ${key}")
      val lv = GroupByAction.convertByteArrayToJsonObject(value)
      val lst = List.empty[String]
      val newKey = GroupByAction
        .reKey(lv
          , groupByColumnList
            .asScala
            .toList
          ,lst)
      val newValue = getValFromJSONMessage(lv, aggregateColumnList.asScala.toList.head)
      println(s"newKey = ${newKey}")
      (newKey, newValue)}
    .groupByKey(Serialized.`with`(Serdes.String(), Serdes.String()))
    .aggregate{ () => 0.toString, (k,v,vr: Long) => (v.toString.toLong + vr.toString.toLong).toString }

score 2 · Accepted Answer · answered Sep 08 '18 at 10:19

2

You can only use the {} form of method call to pass a single parameter, which is treated as a block. You need to pass two parameters, so use () instead:

aggregate( () => 0.toString, (k,v,vr: Long) => (v.toString.toLong + vr.toString.toLong).toString )

answered Sep 08 '18 at 10:19

Alexey Romanov

167,066
35
309
487

Getting unspecified value parameter scala Kafka

1 Answers1