Cannot chain streaming functions

[fbiville]

By chaining, I mean have a stream that is both the output stream of a function and the input stream of another function.

You can observe the issue while running this script.

The way to observe the problem is the following:

• the Liiklus client for the output stream of the first function observes data
• while the Liiklus for the last function's output stream does not

[jldec]

processor is crashing

Exception in thread "main" java.lang.IllegalArgumentException: Expected authority at index 2: //
    at java.net.URI.create(URI.java:852)
    at io.grpc.internal.DnsNameResolver.<init>(DnsNameResolver.java:164)
    at io.grpc.internal.DnsNameResolverProvider.newNameResolver(DnsNameResolverProvider.java:58)
    at io.grpc.internal.DnsNameResolverProvider.newNameResolver(DnsNameResolverProvider.java:41)
    at io.grpc.NameResolverRegistry$NameResolverFactory.newNameResolver(NameResolverRegistry.java:154)
    at io.grpc.internal.ManagedChannelImpl.getNameResolver(ManagedChannelImpl.java:696)
    at io.grpc.internal.ManagedChannelImpl.<init>(ManagedChannelImpl.java:578)
    at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:512)
    at io.projectriff.processor.Processor.lambda$indexByAddress$9(Processor.java:218)
    at java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1321)
    at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
    at java.util.stream.DistinctOps$1$2.accept(DistinctOps.java:175)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1556)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
    at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
    at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566)
    at io.projectriff.processor.Processor.indexByAddress(Processor.java:213)
    at io.projectriff.processor.Processor.<init>(Processor.java:165)
    at io.projectriff.processor.Processor.main(Processor.java:125)
Caused by: java.net.URISyntaxException: Expected authority at index 2: //
    at java.net.URI$Parser.fail(URI.java:2848)
    at java.net.URI$Parser.failExpecting(URI.java:2854)
    at java.net.URI$Parser.parseHierarchical(URI.java:3102)
    at java.net.URI$Parser.parse(URI.java:3063)
    at java.net.URI.<init>(URI.java:588)
    at java.net.URI.create(URI.java:850)
    ... 21 more

[jldec]

According to @ericbottard

I think this happens when there is a problem with the provisionner and the Status of the Stream doesn't reflect its address. Specifically https://github.com/projectriff/kafka-provisioner/issues/2

2 Provisioner loses connectivity to kafka?

After some time of inactivity, it seems newly created streams don't get ready.
Killing just the provisioner pod solves the situation, which seems to indicate some connection timeout is at play. https://github.com/projectriff/kafka-provisioner|projectriff/kafka-provisioner

[fbiville]

I edited the initial issue description on how to observe the problem.

[ericbottard]

Need a better way to reproduce this, as the provided script is rather stale wrt master:

• --provider flag needs the service of the provisioner part of a kafka-provider. For a provider named franz, that would typically be franz-kafka-provisioner
• liiklus-client needs to use the internal names of kafka topics created, so for stream named numbers in namespace default, that would typically be default_numbers
• the service created for liiklus (name is wrong here as well) is not exposed to the outside world anymore, so the piece of jsonpath magic would not work. I'd recomment kubefwd or equivalent to expose the service locally

In any case, after some other unrelated hiccups, chaining worked as intented for me. So, need to narrow down on the problem, if any...

[ericbottard]

Here is a better script for running the demo, assuming

• a kafka-provider was created with name franz (i.e. riff streaming kafka-provider create franz --bootstrap-servers=kafkabroker:9092)
• usage of kubefwd (or equivalent) to expose the liiklus service locally using the exact name of the service

[jldec]

Closing, cannot reproduce chaining failure.