[ https://issues.apache.org/jira/browse/KUDU-2116?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Grant Henke resolved KUDU-2116. ------------------------------- Fix Version/s: NA Resolution: Duplicate > kudu client multi thread scanner throw exception:Invalid call sequence ID in > scan request > ----------------------------------------------------------------------------------------- > > Key: KUDU-2116 > URL: https://issues.apache.org/jira/browse/KUDU-2116 > Project: Kudu > Issue Type: Bug > Components: client, java > Affects Versions: 1.4.0 > Environment: java8 > Development environment win 7 > kudu server version is 'kudu 1.4.0-cdh5.12.0' > kudu client version is '1.4.0' > Reporter: WangBo > Assignee: Todd Lipcon > Priority: Major > Fix For: NA > > > I want to use kudu-client to fetch data.To improve efficiency,I use multi > thread to do this.Use KuduScanToken.intoScanner to get KuduScanner,then use > KuduScanner to fetch data.In single thread environment,it's ok.but when using > multi thread,every thread holds a kuduScanner to fetch data,exception occurs > as below: > {panel:title=My title} > org.apache.kudu.client.NonRecoverableException: Invalid call sequence ID in > scan request > at > org.apache.kudu.client.TabletClient.dispatchTSErrorOrReturnException(TabletClient.java:526) > at > org.apache.kudu.client.TabletClient.messageReceived(TabletClient.java:456) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) > at > org.apache.kudu.client.TabletClient.handleUpstream(TabletClient.java:603) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) > at > org.apache.kudu.client.shaded.org.jboss.netty.handler.timeout.ReadTimeoutHandler.messageReceived(ReadTimeoutHandler.java:184) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) > at > org.apache.kudu.client.shaded.org.jboss.netty.handler.codec.oneone.OneToOneDecoder.handleUpstream(OneToOneDecoder.java:70) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) > at > org.apache.kudu.client.shaded.org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462) > at > org.apache.kudu.client.shaded.org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443) > at > org.apache.kudu.client.shaded.org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:303) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:337) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89) > at > org.apache.kudu.client.shaded.org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) > at > org.apache.kudu.client.shaded.org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) > at > org.apache.kudu.client.shaded.org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) > at > java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) > at > java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) > at java.lang.Thread.run(Thread.java:745) > {panel} > the client code as below: > {code:java} > public class ScannerTest { > ExecutorService pool = Executors.newFixedThreadPool(8); > @Test > public void scan() throws Exception { > long begin = System.currentTimeMillis(); > String tableName = "impala::test.test_table"; > KuduClient kuduClient = getKuduClient(); > KuduTable kuduTable = kuduClient.openTable(tableName); > List<KuduScanToken> kuduScanTokens = > kuduClient.newScanTokenBuilder(kuduTable) > .build(); > List<Future> list = Lists.newArrayList(); > for (int i = 0;i < kuduScanTokens.size();i ++) { > KuduScanToken kuduScanToken = kuduScanTokens.get(i); > KuduScanner kuduScanner = > kuduScanToken.intoScanner(getKuduClient()); > list.add(pool.submit(new Callable<Object>() { > public Object call() { > String name = Thread.currentThread().getName(); > try { > System.out.println("scan start:" + name); > doKuduScan(kuduScanner, name); > return name; > } catch (KuduException e) { > e.printStackTrace(); > return e.getMessage(); > } > } > })); > } > list.forEach(future -> { > try { > System.out.println(future.get()); > } catch (InterruptedException e) { > e.printStackTrace(); > } catch (ExecutionException e) { > e.printStackTrace(); > } > }); > pool.shutdown(); > } > public void doKuduScan(KuduScanner kuduScanner, String threadName) throws > KuduException { > while (kuduScanner.hasMoreRows()) { > RowResultIterator results = kuduScanner.nextRows(); > System.out.println(threadName + ",num=" + results.getNumRows()); > while (results.hasNext()) { > RowResult rowResult = results.next(); > } > } > } > public static KuduClient getKuduClient() { > KuduClient kuduClient = new > KuduClient.KuduClientBuilder(hostName).build(); > return kuduClient; > } > } > {code} > the more thread num is,the more easily case occurs -- This message was sent by Atlassian Jira (v8.3.4#803005)