You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by "ROHIT RAJKUMAR GARG (JIRA)" <ji...@apache.org> on 2017/05/11 10:47:04 UTC
[jira] [Commented] (APEXCORE-718) How to Use one Kafka input
operator and Feed to multiple operator
[ https://issues.apache.org/jira/browse/APEXCORE-718?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16006224#comment-16006224 ]
ROHIT RAJKUMAR GARG commented on APEXCORE-718:
----------------------------------------------
dag.addStream("kafkaData", kafkainput.outputPort , consumptionKeyValueCreator.input, requestCountKeyValueCreator.input).setLocality(Locality.CONTAINER_LOCAL);
dag.addStream("JsonData", consumptionKeyValueCreator.output , consumptionOperator.data).setLocality(Locality.CONTAINER_LOCAL);
dag.addStream("JsonData1", requestCountKeyValueCreator.output , countOperator.data).setLocality(Locality.CONTAINER_LOCAL);
dag.addStream("ConsumptionData1", countOperator.sum , countMapOutput.input).setLocality(Locality.CONTAINER_LOCAL);
---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
java.lang.IllegalArgumentException: Port input already connected to stream LogicalPlan.StreamMeta[id=MapData]
at com.datatorrent.stram.plan.logical.LogicalPlan$StreamMeta.addSink(LogicalPlan.java:548)
at com.datatorrent.stram.plan.logical.LogicalPlan.addStream(LogicalPlan.java:1429)
at com.datatorrent.stram.plan.logical.LogicalPlan.addStream(LogicalPlan.java:1480)
at com.datatorrent.stram.plan.logical.LogicalPlan.addStream(LogicalPlan.java:125)
at org.jio.media.Application.populateDAG(Application.java:68)
at com.datatorrent.stram.plan.logical.LogicalPlanConfiguration.prepareDAG(LogicalPlanConfiguration.java:2263)
at com.datatorrent.stram.plan.logical.LogicalPlanConfiguration.createFromStreamingApplication(LogicalPlanConfiguration.java:2123)
at org.apache.apex.engine.util.StreamingAppFactory.createApp(StreamingAppFactory.java:47)
at com.datatorrent.stram.client.StramAppLauncher$1.createApp(StramAppLauncher.java:473)
at com.datatorrent.stram.client.StramAppLauncher.runLocal(StramAppLauncher.java:518)
at com.datatorrent.stram.cli.ApexCli$LaunchCommand.execute(ApexCli.java:2143)
at com.datatorrent.stram.cli.ApexCli.launchAppPackage(ApexCli.java:3561)
at com.datatorrent.stram.cli.ApexCli.access$7400(ApexCli.java:153)
at com.datatorrent.stram.cli.ApexCli$LaunchCommand.execute(ApexCli.java:1980)
at com.datatorrent.stram.cli.ApexCli$3.run(ApexCli.java:1539)
> How to Use one Kafka input operator and Feed to multiple operator
> ------------------------------------------------------------------
>
> Key: APEXCORE-718
> URL: https://issues.apache.org/jira/browse/APEXCORE-718
> Project: Apache Apex Core
> Issue Type: Question
> Reporter: ROHIT RAJKUMAR GARG
>
> I want to read from a topic using the kafka input operator and then feed it to multiple operator . in short it will be a sort of branch
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)