Kafka Streams：第n个事件的动作

我正在尝试找到在Kafka Streams中对第n个事件执行操作的最佳方法。

我的情况：我有一个带有一些Events的输入流。我必须通过eventType == login过滤它们，并在每次第n次登录（假设是第五次）时将同一accountId将此Event发送到输出流。

经过一些调查和不同的尝试，我有了下面的代码版本（我使用的是Kotlin）。

data class Event(
    val payload: Any = {},
    val accountId: String,
    val eventType: String = ""
)

// intermediate class to keep the key and value of the original event
data class LoginEvent(
    val eventKey: String,
    val eventValue: Event
)

fun process() {
        val userLoginsStoreBuilder = Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("logins"),
            Serdes.String(),
            Serdes.Integer()
        )
        val streamsBuilder = StreamsBuilder().addStateStore(userCheckInsStoreBuilder)
        val inputStream = streamsBuilder.stream<String, String>(inputTopic)

        inputStream.map { key, event ->
            KeyValue(key, json.readValue<Event>(event))
        }.filter { _, event -> event.eventType == "login" }
             .map { key, event -> KeyValue(event.accountId, LoginEvent(key, event)) }
             .transform(
                    UserLoginsTransformer("logins", 5),
                    "logins"
                )
             .filter { _, value -> value }
             .map { key, _ -> KeyValue(key.eventKey, json.writeValueAsString(key.eventValue)) }
             .to("fifth_login", Produced.with(Serdes.String(), Serdes.String()))

        ...
    }

class UserLoginsTransformer(private val storeName: String, private val loginsThreshold: Int = 5) :
    TransformerSupplier<String, CheckInEvent, KeyValue< LoginEvent, Boolean>> {

    override fun get(): Transformer<String, LoginEvent, KeyValue< LoginEvent, Boolean>> {
        return object : Transformer<String, LoginEvent, KeyValue< LoginEvent, Boolean>> {
            private lateinit var store: KeyValueStore<String, Int>

            @Suppress("UNCHECKED_CAST")
            override fun init(context: ProcessorContext) {
                store = context.getStateStore(storeName) as KeyValueStore<String, Int>
            }

            override fun transform(key: String, value: LoginEvent): KeyValue< LoginEvent, Boolean> {
                val counter = (store.get(key) ?: 0) + 1
                return if (counter == loginsThreshold) {
                    store.delete(key)
                    KeyValue(value, true)
                } else {
                    store.put(key, counter)
                    KeyValue(value, false)
                }
            }

            override fun close() {
            }
        }
    }
}

我最大的担心是transform我的情况下函数不是线程安全的。我已经检查了我的案例中使用的KV存储的实现，这是RocksDB存储（非事务性），因此该值可能会在读取和比较之间更新，并且错误的事件将发送到输出。

我的其他想法：

1. 可以将物化视图用作没有转换器的存储，但是我坚持实现。
2. 创建一个将使用TransactionalRocksDB的自定义持久性KV存储（不确定是否值得）。
3. 创建一个自定义的持久性KV存储，该存储将在内部使用ConcurrentHashMap（在我们期望有许多用户的情况下，这可能会导致高内存消耗）。

还有一点要注意：我正在使用Spring Cloud Stream，所以也许该框架针对我的情况有一个内置的解决方案，但是我没有找到它。

我将不胜感激任何建议。提前致谢。