Change netty reply timeout to 2s and run a periodic cleanup task to timeout outstanding requests

Change-Id: Ie0381b9371bfd8a3d680872bfc5dce54a19aaca2

@@ -84,6 +84,7 @@ import java.util.function.Consumer;
 @Service
 @Service
 public class NettyMessagingManager implements MessagingService {
 public class NettyMessagingManager implements MessagingService {
 
 
+    private static final int REPLY_TIME_OUT_SEC = 2;
     private static final short MIN_KS_LENGTH = 6;
     private static final short MIN_KS_LENGTH = 6;
 
 
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Logger log = LoggerFactory.getLogger(getClass());
@@ -96,7 +97,7 @@ public class NettyMessagingManager implements MessagingService {
     private final Map<String, Consumer<InternalMessage>> handlers = new ConcurrentHashMap<>();
     private final Map<String, Consumer<InternalMessage>> handlers = new ConcurrentHashMap<>();
     private final AtomicLong messageIdGenerator = new AtomicLong(0);
     private final AtomicLong messageIdGenerator = new AtomicLong(0);
     private final Cache<Long, Callback> callbacks = CacheBuilder.newBuilder()
     private final Cache<Long, Callback> callbacks = CacheBuilder.newBuilder()
-            .expireAfterWrite(10, TimeUnit.SECONDS)
+            .expireAfterWrite(REPLY_TIME_OUT_SEC, TimeUnit.SECONDS)
             .removalListener(new RemovalListener<Long, Callback>() {
             .removalListener(new RemovalListener<Long, Callback>() {
                 @Override
                 @Override
                 public void onRemoval(RemovalNotification<Long, Callback> entry) {
                 public void onRemoval(RemovalNotification<Long, Callback> entry) {
@@ -145,6 +146,7 @@ public class NettyMessagingManager implements MessagingService {
         initEventLoopGroup();
         initEventLoopGroup();
         startAcceptingConnections();
         startAcceptingConnections();
         started.set(true);
         started.set(true);
+        serverGroup.scheduleWithFixedDelay(callbacks::cleanUp, 0, REPLY_TIME_OUT_SEC, TimeUnit.SECONDS);
         log.info("Started");
         log.info("Started");
     }
     }