@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(); }
@Override public AlphaResponse send(TxEvent event) { return (AlphaResponse) send("sendEvent", event); }
LoadBalancedClusterMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { return (AlphaResponse) send("sendEvent", event); } }
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); }
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); }
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);  }