You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by "Aditya (JIRA)" <ji...@apache.org> on 2019/02/20 00:57:00 UTC
[jira] [Updated] (SAMZA-1990) Samza framework should let using the
same system stream as both input and output.
[ https://issues.apache.org/jira/browse/SAMZA-1990?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Aditya updated SAMZA-1990:
--------------------------
Comment: was deleted
(was: [~pmaheshwari]: Could we take a look at this for the table, where input and output tables are the same ?
It currently fails with the below exception:
2019-02-19 08:42:00.947 [main] SamzaSqlApplication [ERROR] SamzaSqlApplication threw exception.
java.lang.IllegalStateException: Duplicate key com.linkedin.samza.sql.schema.EspressoRelSchemaProviderFactory$EspressoRelSchemaProvider@2dfb885e
at java.util.stream.Collectors.lambda$throwingMerger$0(Collectors.java:133)
at java.util.HashMap.merge(HashMap.java:1254)
at java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1320)
at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
at java.util.HashMap$KeySpliterator.forEachRemaining(HashMap.java:1556)
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 org.apache.samza.sql.runner.SamzaSqlApplicationConfig.<init>(SamzaSqlApplicationConfig.java:138)
at org.apache.samza.sql.runner.SamzaSqlApplication.init(SamzaSqlApplication.java:44)
at org.apache.samza.runtime.AbstractApplicationRunner.getExecutionPlan(AbstractApplicationRunner.java:130)
at org.apache.samza.runtime.RemoteApplicationRunner.run(RemoteApplicationRunner.java:72)
at org.apache.samza.sql.runner.SamzaSqlApplicationRunner.run(SamzaSqlApplicationRunner.ja)
> Samza framework should let using the same system stream as both input and output.
> ---------------------------------------------------------------------------------
>
> Key: SAMZA-1990
> URL: https://issues.apache.org/jira/browse/SAMZA-1990
> Project: Samza
> Issue Type: Bug
> Reporter: Aditya
> Priority: Major
>
> Consider the following scenario: A StreamApplication containing two pipelines with the output StreamDescriptor of the first pipeline being the input StreamDescriptor for the second pipeline. With Samza 1.0 changes, this results in the "Serde for streamId: testavro-SIMPLE1 is already defined. Cannot change it to different serdes." exception. This is an important use-case for Samza-sql. We need to fix this.
> org.apache.samza.config.ConfigException: Could not load ApplicationRunner class org.apache.samza.runtime.LocalApplicationRunner
>
> at org.apache.samza.runtime.ApplicationRunners.getApplicationRunner(ApplicationRunners.java:74)
> at org.apache.samza.sql.runner.SamzaSqlApplicationRunner.<init>(SamzaSqlApplicationRunner.java:73)
> at org.apache.samza.sql.runner.SamzaSqlApplicationRunner.<init>(SamzaSqlApplicationRunner.java:69)
> at org.apache.samza.test.samzasql.TestSamzaSqlEndToEnd.testEndToEndMultiSqlStmtsWithSameSystemStreamAsInputAndOutput(TestSamzaSqlEndToEnd.java:142)
> 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.runners.ParentRunner.run(ParentRunner.java:363)
> at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
> at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
> at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
> at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
> at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
> Caused by: java.lang.reflect.InvocationTargetException
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
> at org.apache.samza.runtime.ApplicationRunners.getApplicationRunner(ApplicationRunners.java:68)
> ... 27 more
> Caused by: java.lang.IllegalArgumentException: Serde for streamId: testavro-SIMPLE1 is already defined. Cannot change it to different serdes.
> at org.apache.samza.application.descriptors.ApplicationDescriptorImpl.getOrCreateStreamSerdes(ApplicationDescriptorImpl.java:315)
> at org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl.getInputStream(StreamApplicationDescriptorImpl.java:97)
> at org.apache.samza.sql.translator.ScanTranslator.lambda$translate$0(ScanTranslator.java:112)
> at java.util.HashMap.computeIfAbsent(HashMap.java:1118)
> at org.apache.samza.sql.translator.ScanTranslator.translate(ScanTranslator.java:112)
> at org.apache.samza.sql.translator.QueryTranslator$1.visit(QueryTranslator.java:154)
> at org.apache.calcite.rel.core.TableScan.accept(TableScan.java:167)
> at org.apache.calcite.rel.RelShuttleImpl.visitChild(RelShuttleImpl.java:55)
> at org.apache.calcite.rel.RelShuttleImpl.visit(RelShuttleImpl.java:99)
> at org.apache.samza.sql.translator.QueryTranslator$1.visit(QueryTranslator.java:167)
> at org.apache.calcite.rel.logical.LogicalProject.accept(LogicalProject.java:129)
> at org.apache.calcite.rel.RelShuttleImpl.visitChild(RelShuttleImpl.java:55)
> at org.apache.calcite.rel.RelShuttleImpl.visitChildren(RelShuttleImpl.java:69)
> at org.apache.calcite.rel.RelShuttleImpl.visit(RelShuttleImpl.java:131)
> at org.apache.samza.sql.translator.QueryTranslator$1.visit(QueryTranslator.java:146)
> at org.apache.samza.sql.translator.QueryTranslator$1.visit(QueryTranslator.java:137)
> at org.apache.calcite.rel.AbstractRelNode.accept(AbstractRelNode.java:279)
> at org.apache.samza.sql.translator.QueryTranslator.translate(QueryTranslator.java:130)
> at org.apache.samza.sql.runner.SamzaSqlApplication.describe(SamzaSqlApplication.java:74)
> at org.apache.samza.sql.runner.SamzaSqlApplication.describe(SamzaSqlApplication.java:42)
> at org.apache.samza.application.descriptors.StreamApplicationDescriptorImpl.<init>(StreamApplicationDescriptorImpl.java:81)
> at org.apache.samza.application.descriptors.ApplicationDescriptorUtil.getAppDescriptor(ApplicationDescriptorUtil.java:45)
> at org.apache.samza.runtime.LocalApplicationRunner.<init>(LocalApplicationRunner.java:74)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)