### Question:
LoadBalancedClusterMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { return (AlphaResponse) send("sendEvent", event); } LoadBalancedClusterMessageSender(AlphaClusterConfig clusterConfig,
      MessageSerializer serializer,
      MessageDeserializer deserializer,
      ServiceConfig serviceConfig,
      MessageHandler handler,
      int reconnectDelay,
      Tracing tracing);   LoadBalancedClusterMessageSender(MessageSender... messageSenders); @Override void onConnected(); @Override void onDisconnected(); @Override void close(); @Override String target(); @Override AlphaResponse send(TxEvent event); @Override Set<String> send(Set<String> localTxIdSet); @Override String reportMessageToServer(KafkaMessage message); @Override String reportAccidentToServer(AccidentHandling accidentHandling); @Override GrpcConfigAck readConfigFromServer(int type, String category);  }### Answer:
@Test public void resendToAnotherServerOnFailure() throws Exception { messageSender.send(event); int deadPort = killServerReceivedMessage(); messageSender.send(event); messageSender.send(event); assertThat(eventsMap.get(deadPort).size(), is(1)); assertThat(eventsMap.get(deadPort).peek().toString(), is(event.toString())); int livePort = deadPort == 8080 ? 8090 : 8080; assertThat(eventsMap.get(livePort).size(), is(2)); assertThat(eventsMap.get(livePort).peek().toString(), is(event.toString())); startServerOnPort(deadPort); }
@Test(timeout = 1000) public void stopSendingOnInterruption() throws Exception { MessageSender underlying = Mockito.mock(MessageSender.class); doThrow(RuntimeException.class).when(underlying).send(event); final MessageSender messageSender = new LoadBalancedClusterMessageSender(underlying); Thread thread = new Thread(new Runnable() { @Override public void run() { messageSender.send(event); } }); thread.start(); Thread.sleep(300); verify(underlying, times(1)).send(event); thread.interrupt(); thread.join(); }
@Test public void considerFasterServerFirst() throws Exception { messageSender.send(event); messageSender.send(event); messageSender.send(event); messageSender.send(event); assertThat(eventsMap.get(8080).size(), is(3)); assertThat(eventsMap.get(8090).size(), is(1)); }
@Test public void forwardSendResult() { assertThat(messageSender.send(event).aborted(), is(false)); TxEvent rejectEvent = new TxStartedEvent(globalTxId, localTxId, parentTxId, "reject", 0, "", 0, "blah"); assertThat(messageSender.send(rejectEvent).aborted(), is(true)); }
@Test public void blowsUpWhenServerIsInterrupted() throws InterruptedException { Thread thread = new Thread(new Runnable() { @Override public void run() { try { messageSender.send(event); expectFailing(OmegaException.class); } catch (OmegaException e) { assertThat(e.getMessage().endsWith("interruption"), is(true)); } } }); thread.start(); thread.interrupt(); thread.join(); }