Issue with "test from file" feature

[Mathieu Pedrero]

Hi again,

As I manage to build a workaround for the docker image issue I sent to the newsletter some minutes ago, I take the opportunity to also send a mail related to the issue I'm currently struggling with :

I have some trouble using the "test from file" feature from the ui on a basic process I built (A sourc, a simple filter, a sink)

I built a basic model (Java project, depending on the following libraries :

compile mySupplyingAppGroup + ':mySupplyingAppModelc:' + mySupplyingAppVersion
compile 'pl.touk.nussknacker:nussknacker-api_2.12:' + nussknackerVersion
compile 'pl.touk.nussknacker:nussknacker-kafka_2.12:' + nussknackerVersion
compile 'pl.touk.nussknacker:nussknacker-kafka-flink-util_2.12:' + nussknackerVersion
compile 'pl.touk.nussknacker:nussknacker-flink-test-util_2.12:' + nussknackerVersion
compile 'pl.touk.nussknacker:nussknacker-kafka-test-util_2.12:' + nussknackerVersion
compile 'pl.touk.nussknacker:nussknacker-process_2.12:' + nussknackerVersion
compile 'org.apache.flink:flink-streaming-scala_2.12:' + flinkVersion

Where nussknacker version is 0.0.12 and flink version is 1.7.2.

I build a shadowJar to package my model with the following options

shadowJar {
    zip64 true
    relocate 'org.apache.flink', 'shadow.flink'
}

The relocation of the org.apache.flink package was set up fo fix a dependency issue I had previously (But since I had the error when I was using the staging nussknacker image, maybe this relocation isn't needed any more).

I manage to generate test dataset from the tool, and to download the file.

But when I try to test my process feeding it the aforementioned dataset, it fails with the following notification.

In the nussknacker logs, the trace is the following :

08:53:15.481 [nussknacker-ui-akka.actor.default-dispatcher-101] ERROR p.t.n.ui.api.EspErrorToHttp$ - Unknown error: pl.touk.nussknacker.engine.process.runner.FlinkTestMain$.run(pl.touk.nussknacker.engine.ModelData, java.lang.String, pl.touk.nussknacker.engine.api.deployment.TestProcess$TestData, org.apache.flink.configuration.Configuration, scala.Function1)
java.lang.NoSuchMethodException: pl.touk.nussknacker.engine.process.runner.FlinkTestMain$.run(pl.touk.nussknacker.engine.ModelData, java.lang.String, pl.touk.nussknacker.engine.api.deployment.TestProcess$TestData, org.apache.flink.configuration.Configuration, scala.Function1)
    at java.lang.Class.getDeclaredMethod(Class.java:2130) ~[na:1.8.0_252]
    at scala.reflect.runtime.JavaMirrors$JavaMirror.$anonfun$methodToJava$1(JavaMirrors.scala:1329) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.TwoWayCaches$TwoWayCache.$anonfun$toJava$1(TwoWayCaches.scala:61) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.TwoWayCaches$TwoWayCache.toJava(TwoWayCaches.scala:57) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror.methodToJava(JavaMirrors.scala:1322) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaMethodMirror.jmeth$lzycompute(JavaMirrors.scala:365) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaMethodMirror.jmeth(JavaMirrors.scala:365) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaMethodMirror.jinvokeraw(JavaMirrors.scala:369) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaMethodMirror.jinvoke(JavaMirrors.scala:373) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.reflect.runtime.JavaMirrors$JavaMirror$JavaVanillaMethodMirror.apply(JavaMirrors.scala:389) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.engine.util.ReflectUtils$StaticMethodRunner.$anonfun$tryToInvoke$1(ReflectUtils.scala:31) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.engine.util.ThreadUtils$.withThisAsContextClassLoader(ThreadUtils.scala:11) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.engine.util.ReflectUtils$StaticMethodRunner.tryToInvoke(ReflectUtils.scala:30) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.engine.management.FlinkProcessTestRunner.test(FlinkProcessTestRunner.scala:15) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.engine.management.FlinkProcessManager.test(FlinkProcessManager.scala:70) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at pl.touk.nussknacker.ui.process.deployment.ManagementActor$$anonfun$receive$1.$anonfun$applyOrElse$21(ManagementActor.scala:101) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.concurrent.Future.$anonfun$flatMap$1(Future.scala:307) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.concurrent.impl.Promise.$anonfun$transformWith$1(Promise.scala:41) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:64) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:91) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:85) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:91) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415) ~[nussknacker-ui_2.12-0.0.12-assembly.jar:0.0.12]
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289) ~[na:1.8.0_252]
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056) ~[na:1.8.0_252]
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692) ~[na:1.8.0_252]
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157) ~[na:1.8.0_252]

The disturbing fact is that, in my IDEA model project, I can see the mentionned pl.touk.nussknacker.engine.process.runner.FlinkTestMain$ class, containing the run method that can't be found by the nussknacker tool.

I'm still trying to fix this, but any help would be appreciated. Have you any idea about this issue ?

Thanks a lot, and best regards.

--

[Mathieu Pedrero]

I found it ! The package relocation (seems so obvious now that I'm writing the mail...) was the issue !

As i now use a 0.0.12 nussknacker docker image, ther is no flink version conflict any more between my moddel and NK. So I removed the relocation option to build my process, and my test process is runnig now !

Best regards

--

[Arek Burdach]

Hi,

Good to hear that You figure out how to handle this problem.

In the near future we are planning to release new version with change on major part (0.1.0). Meantime you can try preview snapshot. The latest one is available here: https://hub.docker.com/layers/touk/nussknacker/2020-04-22-21-08-staging-f5e9fe479a378051c231292cff76aed8304af952/images/sha256-39e69043717e7e0dfb33de178ef6d2b26a40e3acd3b8ff3f1c466d96703cbb39?context=explore
and on maven repository https://oss.sonatype.org/content/groups/public

You can also preview those changes on https://demo.nussknacker.io
There was a lot of visual changes, we currently support Flink's up to 1.9.x (support for 1.10.x are in progress).

Arek