Hazelcast Jet 0.6.1-使用管道customTransform API编译错误

Question

我在使用管道CustomTransformAPI时遇到了以下编译错误。

private Pipeline buildPipeline2() {
    Pipeline p = Pipeline.create();
    p.drawFrom(Sources.<String, CacheEntry<AuditLogRecord>>remoteMapJournal("cache_AuditLog", getClientConfig(), START_FROM_OLDEST))
          .addTimestamps((v) ->  getTimeStamp(v), 3000)
          .peek()
          .groupingKey((v) -> Tuple2.tuple2(getUserID(v),getTranType(v)))
    .window(WindowDefinition.sliding(SLIDING_WINDOW_LENGTH_MILLIS, SLIDE_STEP_MILLIS))
    //.aggregate(counting(),(winStart, winEnd, key, result) -> String.format("%s %5s %4d", toLocalTime(winEnd), formatKey(key), result))
    .aggregate(counting())
    .map((v)-> getMapKey(v))
    .customTransform("test2", ()-> this)
    .drainTo(Sinks.map("Test"));
    //.drainTo(Sinks.files("c:\\data\\op.txt"));
    return p;
  }

protected boolean tryProcess(int ordinal, Object item) {
    JetEvent jetEvent = (JetEvent)item;
    Object obj = jetEvent.payload();
    tryEmit(ordinal,item);
    return true;
}

incompatible types: inferred type does not conform to upper bound(s)
[ERROR] inferred: java.lang.Object
[ERROR] upper bound(s): java.util.Map.Entry

 .customTransform("test2", ()-> this)
 .drainTo(Sinks.files("c:\\data\\op.txt"));

但是，下面的代码给出了编译错误。

.customTransform("test2", ()-> this)
.drainTo(Sinks.map("Test"));

你能帮我解决这个问题吗？

Answer 1

CustomTransform不是类型安全的。如果无法推断类型参数，则计算结果为Object。但是，sinks.map需要map.entry 。要解决此问题，请向CustomTransform方法添加类型提示：

    .<Map.Entry<YourKeyType, YourValueType>customTransform("test2", ()-> this)
    .drainTo(Sinks.map("Test"));

请记住，如果您的自定义处理器实际上没有返回map.entry，它将在运行时失败。

sinks.files工作是因为它接受对象。