[SAMZA-1821] Fetching systemStream configs causes duplicate key exception

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.0
• Component/s: sql
• Assignee: Unassigned
• Reporter: Weiqing Yang
• Created: 24/Aug/18 06:04
• Updated: 04/Sep/18 23:28
• Resolved: 04/Sep/18 23:28

Description

If users input a samza.sql file including multiple Sql statements which have the same input sources or output destinations, for example,

insert into log.outputStream select * from brooklin.espresso.elasticsearchEnterpriseAccounts

insert into log.outputstream select sfdcAccountId as key, organizationUrn as name2, description as name3 from brooklin.espresso.elasticsearchEnterpriseAccounts

an exception of "java.lang.IllegalStateException: Duplicate key org.apache.samza.sql.interfaces.SqlIOConfig" is thrown when fetching input/output systemStream configs:

java.lang.IllegalStateException: Duplicate key org.apache.samza.sql.interfaces.SqlIOConfig@1bf0f6f6

at java.util.stream.Collectors.lambda$throwingMerger$0(Collectors.java:133)
 at java.util.HashMap.merge(HashMap.java:1245)
 at java.util.stream.Collectors.lambda$toMap$58(Collectors.java:1320)
 at java.util.stream.ReduceOps$3ReducingSink.accept(ReduceOps.java:169)
 at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
 at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
 at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:270)
 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 org.apache.samza.sql.runner.SamzaSqlApplicationConfig.<init>(SamzaSqlApplicationConfig.java:116)
 at com.linkedin.samza.sql.SamzaLiSqlParser.getSamzaSqlResources(SamzaLiSqlParser.java:58)
 at com.linkedin.samza.sql.SamzaLiSqlParser.parseSqlResources(SamzaLiSqlParser.java:51)
 at com.linkedin.samza.sql.SamzaLiSqlParserTest.testParseSqlResources_mixedKinds(SamzaLiSqlParserTest.java:60)
 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.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:108)
 at org.testng.internal.Invoker.invokeMethod(Invoker.java:661)
 at org.testng.internal.Invoker.invokeTestMethod(Invoker.java:869)
 at org.testng.internal.Invoker.invokeTestMethods(Invoker.java:1193)
 at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:126)
 at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:109)
 at org.testng.TestRunner.privateRun(TestRunner.java:744)
 at org.testng.TestRunner.run(TestRunner.java:602)
 at org.testng.SuiteRunner.runTest(SuiteRunner.java:380)
 at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:375)
 at org.testng.SuiteRunner.privateRun(SuiteRunner.java:340)
 at org.testng.SuiteRunner.run(SuiteRunner.java:289)
 at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
 at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:86)
 at org.testng.TestNG.runSuitesSequentially(TestNG.java:1301)
 at org.testng.TestNG.runSuitesLocally(TestNG.java:1226)
 at org.testng.TestNG.runSuites(TestNG.java:1144)
 at org.testng.TestNG.run(TestNG.java:1115)
 at org.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:72)
 at org.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:123)

 

Issue Links

• links to: GitHub Pull Request #616