这是最好的方法吗？或者我遗漏了什么可以让这更容易/更好的东西？
我认为您有一个使用本机java类的简单解决方案，可以有效地将streams应用程序与您的代码连接起来。。。为了简单，有很多话要说！我能看到的唯一缺点是，如果事件率太高，用户缓存可能会超出内存大小。此外，如果您需要容错，流应用程序将在出现故障时在另一个应用程序示例上重建状态存储的内容。但如果这不是问题，那就去做吧！
但是，就如何在streams应用程序上下文中实现这一点而言，您可以做一些调整：
定义要支持的用户查询的粒度。几分钟？几秒钟？为了辩论，我们说几分钟吧。根据这个粒度设置流的窗口。
定义一个累加器，类似于您现有的累加器，它将接受一个用户记录并将其添加到列表中。有点像 UserRecordGroup 有一个 List 的 UserRecord ，以及一种方法 add(UserEvent evt) 将附加一个 UserRecord 到 List .
然后，您可以构建您的streams应用程序，如：

KStream<String, Message> inboundStream = streamsBuilder.stream("incoming.topic");
 Materialized<String, UserRecordGroup, WindowStore<Bytes, byte[]>> userStore =
 Materialized.<String, UserRecordGroup, WindowStore<Bytes,byte[]>>as("user.messages")
  .withValueSerde(/* your serializers here */);

KTable<String, MessageCache> messageTable = inboundStream
  .filter(this::userExists)
  .peek(this::recordInboundMessage)
  .map(this::markMessage)       // add sequence/timestamp
  .groupByKey()
  .windowedBy(TimeWindows.of(ONE_MINUTE_IN_MS))
  .aggregate(UserRecordGroup::new,
            (key, value, agg) -> { agg.add(value); },
             userStore);

最后，当您想提供来自商店的查询时，您可以

ReadOnlyWindowStore<Integer, UserRecordGroup> store =
   streams.store("user.messages", QueryableStoreTypes.windowStore());
WindowStoreIterator<UserRecordGroup> windowIterator = 
     store.fetch(pathHash, startTimestamp, endTimeStamp);

迭代器将包含不同窗口的所有记录列表；将这些列表合并在一起，就可以得到starttimestamp和endtimestamp之间的用户活动描述。