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();  }
@Test public void testRoundRobinBackoffInitialFailure() throws EventDeliveryException { 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); 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())); } hosts.get(1).setFailed(); for (int i = 0; i < 3; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } hosts.get(1).setOK(); for (int i = 0; i < 3; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals(1 + 2 + 1, hosts.get(0).getAppendCount()); Assert.assertEquals(1, hosts.get(1).getAppendCount()); Assert.assertEquals(1 + 1 + 2, hosts.get(2).getAppendCount()); }
@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()); }
@Test public void testRoundRobinBackoffFailureRecovery() throws EventDeliveryException, InterruptedException { 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("recovery test".getBytes())); } hosts.get(1).setOK(); Thread.sleep(3000); int numEvents = 60; for (int i = 0; i < numEvents; i++) { c.append(EventBuilder.withBody("testing".getBytes())); } Assert.assertEquals(2 + (numEvents / 3), hosts.get(0).getAppendCount()); Assert.assertEquals(0 + (numEvents / 3), hosts.get(1).getAppendCount()); Assert.assertEquals(1 + (numEvents / 3), hosts.get(2).getAppendCount()); }
@Test public void testRandomBackoff() 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); 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", "random"); p.put("backoff", "true"); hosts.get(0).setFailed(); hosts.get(2).setFailed(); RpcClient c = RpcClientFactory.getInstance(p); Assert.assertTrue(c instanceof LoadBalancingRpcClient); for (int i = 0; i < 50; i++) { c.append(EventBuilder.withBody(("test" + String.valueOf(i)).getBytes())); } Assert.assertEquals(50, hosts.get(1).getAppendCount()); Assert.assertEquals(0, hosts.get(0).getAppendCount()); Assert.assertEquals(0, hosts.get(2).getAppendCount()); hosts.get(0).setOK(); hosts.get(1).setFailed(); try { c.append(EventBuilder.withBody("shouldfail".getBytes())); Assert.fail("Expected EventDeliveryException"); } catch (EventDeliveryException e) { } Thread.sleep(2500); for (int i = 0; i < 50; i++) { c.append(EventBuilder.withBody(("test" + String.valueOf(i)).getBytes())); } Assert.assertEquals(50, hosts.get(0).getAppendCount()); Assert.assertEquals(50, hosts.get(1).getAppendCount()); Assert.assertEquals(0, hosts.get(2).getAppendCount()); }