@Test(expected=IOException.class) public void connectionRetryTimeoutTest() throws Exception { Learner learner = new TimeoutLearner(); learner.self = new QuorumPeer(); learner.self.setTickTime(2000); learner.self.setInitLimit(5); learner.self.setSyncLimit(2); InetSocketAddress addr = new InetSocketAddress(1111); learner.connectToLeader(addr, ""); }
protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception { this.sock = createSocket(); int initLimitTime = self.tickTime * self.initLimit; int remainingInitLimitTime = initLimitTime; long startNanoTime = nanoTime(); for (int tries = 0; tries < 5; tries++) { try { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 0) { LOG.error("initLimit exceeded on retries."); throw new IOException("initLimit exceeded on retries."); } sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime)); if (self.isSslQuorum()) { ((SSLSocket) sock).startHandshake(); } sock.setTcpNoDelay(nodelay); break; } catch (IOException e) { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 1000) { LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else if (tries >= 4) { LOG.error("Unexpected exception, retries exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else { LOG.warn("Unexpected exception, tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); this.sock = createSocket(); } } Thread.sleep(1000); } self.authLearner.authenticate(sock, hostname); leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream())); bufferedOutput = new BufferedOutputStream(sock.getOutputStream()); leaderOs = BinaryOutputArchive.getArchive(bufferedOutput); }
Learner { protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception { this.sock = createSocket(); int initLimitTime = self.tickTime * self.initLimit; int remainingInitLimitTime = initLimitTime; long startNanoTime = nanoTime(); for (int tries = 0; tries < 5; tries++) { try { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 0) { LOG.error("initLimit exceeded on retries."); throw new IOException("initLimit exceeded on retries."); } sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime)); if (self.isSslQuorum()) { ((SSLSocket) sock).startHandshake(); } sock.setTcpNoDelay(nodelay); break; } catch (IOException e) { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 1000) { LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else if (tries >= 4) { LOG.error("Unexpected exception, retries exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else { LOG.warn("Unexpected exception, tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); this.sock = createSocket(); } } Thread.sleep(1000); } self.authLearner.authenticate(sock, hostname); leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream())); bufferedOutput = new BufferedOutputStream(sock.getOutputStream()); leaderOs = BinaryOutputArchive.getArchive(bufferedOutput); } }
Learner { protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception { this.sock = createSocket(); int initLimitTime = self.tickTime * self.initLimit; int remainingInitLimitTime = initLimitTime; long startNanoTime = nanoTime(); for (int tries = 0; tries < 5; tries++) { try { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 0) { LOG.error("initLimit exceeded on retries."); throw new IOException("initLimit exceeded on retries."); } sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime)); if (self.isSslQuorum()) { ((SSLSocket) sock).startHandshake(); } sock.setTcpNoDelay(nodelay); break; } catch (IOException e) { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 1000) { LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else if (tries >= 4) { LOG.error("Unexpected exception, retries exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else { LOG.warn("Unexpected exception, tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); this.sock = createSocket(); } } Thread.sleep(1000); } self.authLearner.authenticate(sock, hostname); leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream())); bufferedOutput = new BufferedOutputStream(sock.getOutputStream()); leaderOs = BinaryOutputArchive.getArchive(bufferedOutput); }  }
Learner { protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception { this.sock = createSocket(); int initLimitTime = self.tickTime * self.initLimit; int remainingInitLimitTime = initLimitTime; long startNanoTime = nanoTime(); for (int tries = 0; tries < 5; tries++) { try { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 0) { LOG.error("initLimit exceeded on retries."); throw new IOException("initLimit exceeded on retries."); } sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime)); if (self.isSslQuorum()) { ((SSLSocket) sock).startHandshake(); } sock.setTcpNoDelay(nodelay); break; } catch (IOException e) { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 1000) { LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else if (tries >= 4) { LOG.error("Unexpected exception, retries exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else { LOG.warn("Unexpected exception, tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); this.sock = createSocket(); } } Thread.sleep(1000); } self.authLearner.authenticate(sock, hostname); leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream())); bufferedOutput = new BufferedOutputStream(sock.getOutputStream()); leaderOs = BinaryOutputArchive.getArchive(bufferedOutput); }  Socket getSocket(); int getPendingRevalidationsCount(); void shutdown(); }
Learner { protected void connectToLeader(InetSocketAddress addr, String hostname) throws IOException, InterruptedException, X509Exception { this.sock = createSocket(); int initLimitTime = self.tickTime * self.initLimit; int remainingInitLimitTime = initLimitTime; long startNanoTime = nanoTime(); for (int tries = 0; tries < 5; tries++) { try { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 0) { LOG.error("initLimit exceeded on retries."); throw new IOException("initLimit exceeded on retries."); } sockConnect(sock, addr, Math.min(self.tickTime * self.syncLimit, remainingInitLimitTime)); if (self.isSslQuorum()) { ((SSLSocket) sock).startHandshake(); } sock.setTcpNoDelay(nodelay); break; } catch (IOException e) { remainingInitLimitTime = initLimitTime - (int)((nanoTime() - startNanoTime) / 1000000); if (remainingInitLimitTime <= 1000) { LOG.error("Unexpected exception, initLimit exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else if (tries >= 4) { LOG.error("Unexpected exception, retries exceeded. tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); throw e; } else { LOG.warn("Unexpected exception, tries=" + tries + ", remaining init limit=" + remainingInitLimitTime + ", connecting to " + addr,e); this.sock = createSocket(); } } Thread.sleep(1000); } self.authLearner.authenticate(sock, hostname); leaderIs = BinaryInputArchive.getArchive(new BufferedInputStream( sock.getInputStream())); bufferedOutput = new BufferedOutputStream(sock.getOutputStream()); leaderOs = BinaryOutputArchive.getArchive(bufferedOutput); }  Socket getSocket(); int getPendingRevalidationsCount(); void shutdown();  }