Github user roshannaik commented on a diff in the pull request:
https://github.com/apache/storm/pull/2241#discussion_r158930144
--- Diff: storm-client/src/jvm/org/apache/storm/messaging/netty/Client.java
---
@@ -297,31 +333,38 @@ public void send(Iterator<TaskMessage> msgs) {
dropMessages(msgs);
return;
}
-
- synchronized (writeLock) {
+ try {
while (msgs.hasNext()) {
TaskMessage message = msgs.next();
- MessageBatch full = batcher.add(message);
- if(full != null){
- flushMessages(channel, full);
+ MessageBatch batch = batcher.add(message);
+ if (batch != null) {
+ writeMessage(channel, batch);
}
}
+ MessageBatch batch = batcher.drain();
+ if (batch != null) {
+ writeMessage(channel, batch);
+ }
+ } catch (IOException e) {
+ dropMessages(msgs);
--- End diff --
The writeMessage() is the only one that throws that IOException ( after
checking if !channel.isConnected()). I think I should add a log msg also.
---
