@Test public void testRoundRobinBackoffIncreasingBackoffs() throws Exception { Properties p = new Properties(); List<LoadBalancedAvroHandler> hosts = new ArrayList<LoadBalancedAvroHandler>(); List<Server> servers = new ArrayList<Server>(); StringBuilder hostList = new StringBuilder(""); for (int i = 0; i < 3; i++) { LoadBalancedAvroHandler s = new LoadBalancedAvroHandler(); hosts.add(s); if(i == 1) { s.setFailed(); } Server srv = RpcTestUtils.startServer(s); servers.add(srv); String name = "h" + i; p.put("hosts." + name, "127.0.0.1:" + srv.getPort()); hostList.append(name).append(" "); } p.put("hosts", hostList.toString().trim()); p.put("client.type", "default_loadbalance"); p.put("host-selector", "round_robin"); p.put("backoff", "true"); RpcClient c = RpcClientFactory.getInstance(p); Assert.assertTrue(c instanceof LoadBalancingRpcClient); for (int i = 0; i < 3; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals(0, hosts.get(1).getAppendCount()); Thread.sleep(2100); for (int i = 0; i < 3; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals(0, hosts.get(1).getAppendCount()); hosts.get(1).setOK(); Thread.sleep(2100); for (int i = 0; i < 3; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals(0, hosts.get(1).getAppendCount()); Thread.sleep(2500); int numEvents = 60; for (int i = 0; i < numEvents; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals( 2 + 2 + 1 + (numEvents/3), hosts.get(0).getAppendCount()); Assert.assertEquals((numEvents/3), hosts.get(1).getAppendCount()); Assert.assertEquals(1 + 1 + 2 + (numEvents/3), hosts.get(2).getAppendCount()); }
@Override public void append(Event event) throws EventDeliveryException { throwIfClosed(); boolean eventSent = false; Iterator<HostInfo> it = selector.createHostIterator(); while (it.hasNext()) { HostInfo host = it.next(); try { RpcClient client = getClient(host); client.append(event); eventSent = true; break; } catch (Exception ex) { selector.informFailure(host); LOGGER.warn("Failed to send event to host " + host, ex); } } if (!eventSent) { throw new EventDeliveryException("Unable to send event to any host"); } }
LoadBalancingRpcClient extends AbstractRpcClient { @Override public void append(Event event) throws EventDeliveryException { throwIfClosed(); boolean eventSent = false; Iterator<HostInfo> it = selector.createHostIterator(); while (it.hasNext()) { HostInfo host = it.next(); try { RpcClient client = getClient(host); client.append(event); eventSent = true; break; } catch (Exception ex) { selector.informFailure(host); LOGGER.warn("Failed to send event to host " + host, ex); } } if (!eventSent) { throw new EventDeliveryException("Unable to send event to any host"); } } }
LoadBalancingRpcClient extends AbstractRpcClient { @Override public void append(Event event) throws EventDeliveryException { throwIfClosed(); boolean eventSent = false; Iterator<HostInfo> it = selector.createHostIterator(); while (it.hasNext()) { HostInfo host = it.next(); try { RpcClient client = getClient(host); client.append(event); eventSent = true; break; } catch (Exception ex) { selector.informFailure(host); LOGGER.warn("Failed to send event to host " + host, ex); } } if (!eventSent) { throw new EventDeliveryException("Unable to send event to any host"); } }  }
LoadBalancingRpcClient extends AbstractRpcClient { @Override public void append(Event event) throws EventDeliveryException { throwIfClosed(); boolean eventSent = false; Iterator<HostInfo> it = selector.createHostIterator(); while (it.hasNext()) { HostInfo host = it.next(); try { RpcClient client = getClient(host); client.append(event); eventSent = true; break; } catch (Exception ex) { selector.informFailure(host); LOGGER.warn("Failed to send event to host " + host, ex); } } if (!eventSent) { throw new EventDeliveryException("Unable to send event to any host"); } }  @Override void append(Event event); @Override void appendBatch(List<Event> events); @Override boolean isActive(); @Override void close(); }
LoadBalancingRpcClient extends AbstractRpcClient { @Override public void append(Event event) throws EventDeliveryException { throwIfClosed(); boolean eventSent = false; Iterator<HostInfo> it = selector.createHostIterator(); while (it.hasNext()) { HostInfo host = it.next(); try { RpcClient client = getClient(host); client.append(event); eventSent = true; break; } catch (Exception ex) { selector.informFailure(host); LOGGER.warn("Failed to send event to host " + host, ex); } } if (!eventSent) { throw new EventDeliveryException("Unable to send event to any host"); } }  @Override void append(Event event); @Override void appendBatch(List<Event> events); @Override boolean isActive(); @Override void close();  }