Exception thrown while trying to connect to controller.

[shiveshr]

Problem description

        at io.pravega.client.stream.impl.ControllerResolverFactory.lambda$newNameResolver$103(ControllerResolverFactory.java:68)
        at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
        at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
        at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
        at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
        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:499)
        at io.pravega.client.stream.impl.ControllerResolverFactory.newNameResolver(ControllerResolverFactory.java:69)
        at io.grpc.internal.ManagedChannelImpl.getNameResolver(ManagedChannelImpl.java:440)
        at io.grpc.internal.ManagedChannelImpl.<init>(ManagedChannelImpl.java:394)
        at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:350)
        at io.pravega.client.stream.impl.ControllerImpl.<init>(ControllerImpl.java:140)
        at io.pravega.client.stream.impl.ControllerImpl.<init>(ControllerImpl.java:114)
        at io.pravega.client.admin.impl.StreamManagerImpl.<init>(StreamManagerImpl.java:35)
        at io.pravega.client.admin.StreamManager.create(StreamManager.java:28)
        at io.pravega.connectors.flink.utils.SetupUtils.createTestStream(SetupUtils.java:126)

Problem location
SetupUtils.java

In SetupUtils getControllerUri method:

        return URI.create("tcp://" + this.inProcPravegaCluster.getControllerURI());
    }

It expects inProcPravegaCluster.getControllerUri to be only IP:port value.
However, "tcp://" prefix is already added by inprocPravegaCluster
This causes uri to look like "tcp://tcp:// ..."

There is also a bug in InProcPravegaCluster where it adds ports[0] twice whereby making the return value look like:
"tcp://localhost:ports[0], localhost:ports[0], localhost:ports[1] ..."
So even though flink connector tests start only one controller, they get a comma separated value of
"tcp://localhost:<ports[0]>, localhost:<ports[0]>". Though this is harmless.

Suggestions for an improvement
Fix the issue in setupUtils and InProcPravegaCluster.
InprocPravegaCluster should not include ports[0] twice..
SetupUtils should not add tcp:// as it is already added to the returned value..

[tzulitai]

I'll provide a fix for this as part of working on #42.