I am trying to figure out if I can work with Kotlin and Spark, and use the former's data classes instead of Scala's case classes.

I have the following data class:

data class Transaction(var context: String = "", var epoch: Long = -1L, var items: HashSet<String> = HashSet()) :
    Serializable {
    companion object {
        @JvmStatic
        private val serialVersionUID = 1L
    }
}

And the relevant part of the main routine looks like this:

val transactionEncoder = Encoders.bean(Transaction::class.java)
val transactions = inputDataset
    .groupByKey(KeyExtractor(), KeyExtractor.getKeyEncoder())
    .mapGroups(TransactionCreator(), transactionEncoder)
    .collectAsList()

transactions.forEach { println("collected Transaction=$it") }

With TransactionCreator defined as:

class TransactionCreator : MapGroupsFunction<Tuple2<String, Timestamp>, Row, Transaction> {
    companion object {
        @JvmStatic
        private val serialVersionUID = 1L
    }

    override fun call(key: Tuple2<String, Timestamp>, values: MutableIterator<Row>): Transaction {
        val seq = generateSequence { if (values.hasNext()) values.next().getString(2) else null }
        val items = seq.toCollection(HashSet())
        return Transaction(key._1, key._2.time, items).also { println("inside call Transaction=$it") }
    }
}

However, I think I'm running into some sort of serialization problem, because the set ends up empty after collection. I see the following output:

inside call Transaction=Transaction(context=context1, epoch=1000, items=[c])
inside call Transaction=Transaction(context=context1, epoch=0, items=[a, b])
collected Transaction=Transaction(context=context1, epoch=0, items=[])
collected Transaction=Transaction(context=context1, epoch=1000, items=[])

I've tried a custom KryoRegistrator to see if it was a problem with Kotlin's HashSet:

class MyRegistrator : KryoRegistrator {
    override fun registerClasses(kryo: Kryo) {
        kryo.register(HashSet::class.java, JavaSerializer()) // kotlin's HashSet
    }
}

But it doesn't seem to help. Any other ideas?

Full code here.