bedrock-coherence 12.2.1 - streams and remote collectors may not work

[kosmaty]

I wanted to try new Coherence Stream API using bedrock-coherence. I've created a test case (based on AbstractCoherenceClusterBuilderTest.shouldAccessNamedCache()) with simple call to namedCache.stream().collect(RemoteCollectors.counting()):

@Test
public void shouldSupportRemoteCollectors()
{
    final int               CLUSTER_SIZE   = 3;

    AvailablePortIterator   availablePorts = LocalPlatform.get().getAvailablePorts();
    ClusterPort             clusterPort    = ClusterPort.of(new Capture<>(availablePorts));

    CoherenceClusterBuilder builder        = new CoherenceClusterBuilder();

    builder.include(CLUSTER_SIZE, CoherenceClusterMember.class, clusterPort, ClusterName.of("Access"));

    try (CoherenceCluster cluster = builder.build(Console.system()))
    {
        assertThat(invoking(cluster).getClusterSize(), is(CLUSTER_SIZE));
        NamedCache<String, String> namedCache = cluster.getCache("dist-example", String.class, String.class);
        namedCache.put("key", "hello");

        long count = namedCache.stream().collect(RemoteCollectors.counting());
        assertThat(count, is(1L));
    }
    catch (Exception e)
    {
        e.printStackTrace();
        Assert.fail();
    }
}

But it results in throwing ClassNotFoundException. The full stack trace is:

java.lang.RuntimeException: java.util.concurrent.ExecutionException: java.io.IOException: java.lang.ClassNotFoundException: com.tangosol.util.stream.RemoteCollectors$lambda$counting$82a84efb$1$45F7DB6671538DB12EC390FBABF0A8A7
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.remotelyInvoke(CoherenceNamedCache.java:262)
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.aggregate(CoherenceNamedCache.java:376)
    at com.tangosol.internal.util.stream.AbstractPipeline$AggregatorInvoker.invoke(AbstractPipeline.java:387)
    at com.tangosol.internal.util.stream.AbstractPipeline.invoke(AbstractPipeline.java:222)
    at com.tangosol.internal.util.stream.ReferencePipeline.collect(ReferencePipeline.java:344)
    at com.tangosol.internal.util.stream.ReferencePipeline.collect(ReferencePipeline.java:297)
    at com.oracle.bedrock.runtime.coherence.AbstractCoherenceClusterBuilderTest.shouldSupportRemoteCollectors(AbstractCoherenceClusterBuilderTest.java:498)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
    at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
    at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
    at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:117)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:42)
    at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:262)
    at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:84)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)
Caused by: java.util.concurrent.ExecutionException: java.io.IOException: java.lang.ClassNotFoundException: com.tangosol.util.stream.RemoteCollectors$lambda$counting$82a84efb$1$45F7DB6671538DB12EC390FBABF0A8A7
    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.remotelyInvoke(CoherenceNamedCache.java:228)
    ... 37 more
Caused by: java.io.IOException: java.lang.ClassNotFoundException: com.tangosol.util.stream.RemoteCollectors$lambda$counting$82a84efb$1$45F7DB6671538DB12EC390FBABF0A8A7
    at com.oracle.bedrock.runtime.concurrent.AbstractRemoteChannel$CallableOperation.read(AbstractRemoteChannel.java:621)
    at com.oracle.bedrock.runtime.concurrent.AbstractRemoteChannel$1.run(AbstractRemoteChannel.java:248)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException: com.tangosol.util.stream.RemoteCollectors$lambda$counting$82a84efb$1$45F7DB6671538DB12EC390FBABF0A8A7
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at java.lang.Class.forName0(Native Method)

    at java.lang.Class.forName(Class.java:348)
    at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:628)
    at com.oracle.bedrock.runtime.java.io.ClassLoaderAwareObjectInputStream.resolveClass(ClassLoaderAwareObjectInputStream.java:81)
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1620)
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1521)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1781)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
    at java.io.ObjectInputStream.readArray(ObjectInputStream.java:1714)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1347)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2018)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:1942)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1808)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1353)
    at java.io.ObjectInputStream.readObject(ObjectInputStream.java:373)
    at com.oracle.bedrock.runtime.concurrent.AbstractRemoteChannel$CallableOperation.read(AbstractRemoteChannel.java:605)
    ... 2 more

I suspect that it is not only remote collectors issue, but some more general problem with lambdas.

[brianoliver]

Definitely an issue with Lambda serialization with Bedrock. In the meantime the following work-around works.

Instead of the assertion:

long count = namedCache.stream().collect(RemoteCollectors.counting());
assertThat(count, is(1L));

You can instead do the following:

assertThat(cluster.findAny().get(),
           () -> CacheFactory.getTypedCache("dist-example",
                                            TypeAssertion.withTypes(String.class, String.class))
                 .stream().collect(RemoteCollectors.counting()),
           is(1L));

This submits a lambda containing the expression to be evaluated to any cluster member and asserts that the result is as expected. This is instead of Bedrock attempting to capture the internal state of Coherence and then submit that.

[kosmaty]

Thank you for response. The workaround works fine in this simple case. But I'm affraid that it will not be enough in some more real-life scenario. Anyway, I'm glad to see this issue scheduled for the next release.

[brianoliver]

One of the challenges is "where" things like collectors actually execute. In the case above, it seems clear that they are running "Remotely", but is that remotely across the cluster, or remotely in the Cluster Member that Bedrock is interacting with (or both).

Consider the .stream().forEach(Consumer) case.

Should the Consumer be executed in the Bedrock application, in the Cluster Member that Bedrock is interacting with, or in each of the Cluster Members managing data?

I think the answer should be, in the Bedrock (testing?) application, which follows what the Coherence API does. However this means that Bedrock needs to "stream" the results back from the Cluster Member it is interacting with and then invoke the Consumer locally in your application. A bit challenging but not impossible.

It's entirely possible however that another interpretation of the semantics is required, perhaps that the Consumer is executed in the Cluster Member(s) themselves?

So looking at the stream() and collective pipeline APIs, it's clear that each of the methods may require some careful interpretation as to where something should be executed.

For now it might be challenging to provide an (easy/opaque) universal solution for all methods. Fortunately though some of the recent work we've done with streaming support (over RemoteChannels) will help us should we need to solve things on an each individual basis.

[kosmaty]

There is one more lambda serialization issue that I've found. I tried to use lambda expression as EntryProcessor:

ValueExtractor<InvocableMap.Entry<String, String>, String> ex = e -> e.isPresent() ? e.getValue() : null;
namedCache.invokeAll(entry -> ex.apply(entry));

It works fine when cluster is started using LocalPlatform. But when claster is started using JavaVirtualMachine:

CoherenceCluster cluster = builder.build(JavaVirtualMachine.get(), Console.system())

the LinkageError is thrown:

java.lang.RuntimeException: java.util.concurrent.ExecutionException: java.lang.reflect.InvocationTargetException
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.remotelyInvoke(CoherenceNamedCache.java:262)
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.invokeAll(CoherenceNamedCache.java:360)
    at com.tangosol.util.InvocableMap.invokeAll(InvocableMap.java:85)
    at com.oracle.bedrock.runtime.coherence.ContainerBasedCoherenceClusterBuilderTest.shouldSupportLambdaAsEntryProcessor(ContainerBasedCoherenceClusterBuilderTest.java:100)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
    at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
    at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
    at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
    at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:117)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:42)
    at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:262)
    at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:84)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)
Caused by: java.util.concurrent.ExecutionException: java.lang.reflect.InvocationTargetException
    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1895)
    at com.oracle.bedrock.runtime.coherence.CoherenceNamedCache.remotelyInvoke(CoherenceNamedCache.java:228)
    ... 34 more
Caused by: java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.oracle.bedrock.runtime.concurrent.callable.RemoteMethodInvocation.call(RemoteMethodInvocation.java:133)
    at com.oracle.bedrock.runtime.concurrent.AbstractRemoteChannel$CallableOperation.execute(AbstractRemoteChannel.java:578)
    at com.oracle.bedrock.runtime.concurrent.AbstractRemoteChannel$Executor.run(AbstractRemoteChannel.java:849)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: (Wrapped: Failed request execution for PartitionedCache service on Member(Id=2, Timestamp=2016-12-21 15:44:05.641, Address=10.61.6.12:57380, MachineId=30275, Location=site:gft.com,machine:PLLL0106,process:792, Role=IntellijRtExecutionApplicationAppMain) (Wrapped) no such constructor: com.oracle.bedrock.runtime.coherence.ContainerBasedCoherenceClusterBuilderTest$lambda$shouldSupportLambdaAsEntryProcessor$bee7b61e$1$B3CBA80B23DC783FB40D51EF42FE8654.<init>(ValueExtractor)void/newInvokeSpecial) java.lang.IllegalAccessException: no such constructor: com.oracle.bedrock.runtime.coherence.ContainerBasedCoherenceClusterBuilderTest$lambda$shouldSupportLambdaAsEntryProcessor$bee7b61e$1$B3CBA80B23DC783FB40D51EF42FE8654.<init>(ValueExtractor)void/newInvokeSpecial
    at com.tangosol.util.Base.ensureRuntimeException(Base.java:296)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.Grid.tagException(Grid.CDB:61)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.PartitionedCache.onInvokeFilterRequest(PartitionedCache.CDB:123)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.PartitionedCache$InvokeFilterRequest.run(PartitionedCache.CDB:1)
    at com.tangosol.coherence.component.util.DaemonPool$WrapperTask.run(DaemonPool.CDB:1)
    at com.tangosol.coherence.component.util.DaemonPool$WrapperTask.run(DaemonPool.CDB:32)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.PartitionedService$DaemonPool$WrapperTask.run(PartitionedService.CDB:1)
    at com.tangosol.coherence.component.util.DaemonPool$Daemon.onNotify(DaemonPool.CDB:66)
    at com.tangosol.coherence.component.util.Daemon.run(Daemon.CDB:54)
    ... 1 more
Caused by: java.lang.IllegalAccessException: no such constructor: com.oracle.bedrock.runtime.coherence.ContainerBasedCoherenceClusterBuilderTest$lambda$shouldSupportLambdaAsEntryProcessor$bee7b61e$1$B3CBA80B23DC783FB40D51EF42FE8654.<init>(ValueExtractor)void/newInvokeSpecial
    at java.lang.invoke.MemberName.makeAccessException(MemberName.java:867)
    at java.lang.invoke.MemberName$Factory.resolveOrFail(MemberName.java:1003)
    at java.lang.invoke.MethodHandles$Lookup.resolveOrFail(MethodHandles.java:1381)
    at java.lang.invoke.MethodHandles$Lookup.findConstructor(MethodHandles.java:919)
    at com.tangosol.internal.util.invoke.ClassDefinition.setRemotableClass(ClassDefinition.java:125)
    at com.tangosol.internal.util.invoke.RemotableSupport.realize(RemotableSupport.java:123)
    at com.tangosol.internal.util.invoke.RemoteConstructor.newInstance(RemoteConstructor.java:120)
    at com.tangosol.internal.util.invoke.RemoteConstructor.readResolve(RemoteConstructor.java:231)
    at com.tangosol.util.ExternalizableHelper.realize(ExternalizableHelper.java:4837)
    at com.tangosol.util.ExternalizableHelper.deserializeInternal(ExternalizableHelper.java:3101)
    at com.tangosol.util.ExternalizableHelper.fromBinary(ExternalizableHelper.java:334)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.PartitionedCache$InvokeFilterRequest.deserializeProcessor(PartitionedCache.CDB:7)
    at com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.PartitionedCache.onInvokeFilterRequest(PartitionedCache.CDB:62)
    ... 7 more
Caused by: java.lang.LinkageError: loader constraint violation: when resolving method "com.oracle.bedrock.runtime.coherence.ContainerBasedCoherenceClusterBuilderTest$lambda$shouldSupportLambdaAsEntryProcessor$bee7b61e$1$B3CBA80B23DC783FB40D51EF42FE8654.<init>(Lcom/tangosol/util/ValueExtractor;)V" the class loader (instance of <bootloader>) of the current class, java/lang/Object, and the class loader (instance of com/tangosol/internal/util/invoke/RemotableSupport) for the method's defining class, com/oracle/bedrock/runtime/coherence/ContainerBasedCoherenceClusterBuilderTest$lambda$shouldSupportLambdaAsEntryProcessor$bee7b61e$1$B3CBA80B23DC783FB40D51EF42FE8654, have different Class objects for the type com/tangosol/util/ValueExtractor used in the signature
    at java.lang.invoke.MethodHandleNatives.resolve(Native Method)
    at java.lang.invoke.MemberName$Factory.resolve(MemberName.java:975)
    at java.lang.invoke.MemberName$Factory.resolveOrFail(MemberName.java:1000)
    ... 18 more

But I suspect it might be some problem with Oracle Coherence itself (not Bedrock) since I was facing the same issue when using Littlegrid (in fact, it was the primary reason why I started evaluation of Bedrock).

Please, feel free to move this to another issue if you think it is not strictly connected to the original case of this issue.

[brianoliver]

Let's raise this issue as something separate (ie: now covered by issue #387).

I think it's more likely to do with either the child-first-class loader (the same technique is used by LittleGrid and Bedrock) and/or the use of Lambdas themselves.

Confirmed that this above issue #387 is in fact a Java issue, nothing to do with Coherence, Bedrock or Littlegrid (but there's decent work arounds)