Hazelcast Jet 0.6.1 - Compilation error with Pipeline customTransform API

Question

I am getting following compilation error with Pipeline customTransform API.

Here is sample code for building pipeline:

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;
  }

Here is sample code for tryProcess() method:

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

Here is compilation error.

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

This compiles and runs well with following code.

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

However,the gives compilation error with following code.

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

Will you please help me to resolve this issue ?

Most probably it doesn't work with the map Sink because it expects a complete Map.Entry instance with the key and value. And you're extracting just key from it at the step .map((v)-> getMapKey(v)), so that only JetEvent object goes further into the Sink. And then it thows the error. — Nazar

Oliv Oliv · Accepted Answer · 2018-06-22T07:13:12

customTransform isn't type-safe. If the type parameter can't be inferred, it will evaluate to Object. However, the Sinks.map requires Map.Entry<K, V>. To resolve it, add type hint to the customTransform method:

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

Keep in mind that if your custom processor doesn't actually return Map.Entry, it will fail at runtime.

The Sinks.files works because it takes Object.

Hazelcast Jet 0.6.1 - Compilation error with Pipeline customTransform API

1 Answers