Hi, that's a good question - I'll look more into that and come back to you tomorrow.
-- Alex On Thu, Jan 21, 2016 at 2:23 PM, Vikas Singh (Tech - VS) < vikas.si...@flipkart.com> wrote: > Hi, > > We are facing error similar to > https://issues.apache.org/jira/browse/ZEPPELIN-312. This time it is > happening when for some reason send() function of NotebookSocket gets hung, > blocking the connection and later if the broadcast(String noteId, Message > m) of NoteBookServer is invoked to broadcast to the same connection it will > hang as well and since it is in as synchronized block. Zeppelin server > becomes non responsive since no further broadcasts containing that > connection will take place. > > > For resolving https://issues.apache.org/jira/browse/ZEPPELIN-312. > synchronized block of broadcastAll() function of NotebookServer was removed > and connectedSockets was converted to ConcurrentLinkedQueue. > > I was wondering if similar approach can be taken to fix this issue. by > removing synchronized block of broadcast() method and replacing > noteSocketMap to ConcurrentHashMap. Any ways having network call inside > synchronized block in this case can any time result in blocked threads. > Any comments will be greatly appreciated. > > //To Reproduce this issue (NotebookSocket.java)- > > public void send(String serializeMessage) throws IOException { > if(serializeMessage.startsWith("{\"op\":\"NOTE\"")){ > LOG.info("Writing socket till fails."); > while(true) { > connection.sendMessage(serializeMessage); > } > } > connection.sendMessage(serializeMessage); > } > > > > // Below change seems to fix the above issue (NotebookServer.java) > > -final Map<String, List<NotebookSocket>> noteSocketMap = new > ConcurrentHashMap<String, List<NotebookSocket>>(); > > +final Map<String, List<NotebookSocket>> noteSocketMap = new > ConcurrentHashMap<String, > List<NotebookSocket>>(); > > > private void broadcast(String noteId, Message m) { > > //synchronized (noteSocketMap) { > List<NotebookSocket> socketLists = noteSocketMap.get(noteId); > if (socketLists == null || socketLists.size() == 0) { > return; > } > LOG.info("SEND >> " + m.op); > for (NotebookSocket conn : socketLists) { > try { > conn.send(serializeMessage(m)); > } catch (IOException e) { > LOG.error("socket error- NoteID: " + noteId + "Message: "+ m, e); > } > } > //} > > } > > > Regards > > Vikas Singh >
