@Test public void swallowException_UntilAllSendersConnected() throws Exception { MessageSender underlying1 = Mockito.mock(MessageSender.class); doThrow(RuntimeException.class).when(underlying1).onConnected(); MessageSender underlying2 = Mockito.mock(MessageSender.class); MessageSender sender = new LoadBalancedClusterMessageSender(underlying1, underlying2); sender.onConnected(); verify(underlying1).onConnected(); verify(underlying2).onConnected(); }
@Override public void onConnected() { for (MessageSender sender : senders.keySet()) { try { sender.onConnected(); } catch (Exception e) { LOG.error("Failed connecting to alpha at {}", sender.target(), e); } } }
LoadBalancedClusterMessageSender implements MessageSender { @Override public void onConnected() { for (MessageSender sender : senders.keySet()) { try { sender.onConnected(); } catch (Exception e) { LOG.error("Failed connecting to alpha at {}", sender.target(), e); } } } }
LoadBalancedClusterMessageSender implements MessageSender { @Override public void onConnected() { for (MessageSender sender : senders.keySet()) { try { sender.onConnected(); } catch (Exception e) { LOG.error("Failed connecting to alpha at {}", sender.target(), e); } } } LoadBalancedClusterMessageSender(AlphaClusterConfig clusterConfig,
      MessageSerializer serializer,
      MessageDeserializer deserializer,
      ServiceConfig serviceConfig,
      MessageHandler handler,
      int reconnectDelay,
      Tracing tracing);   LoadBalancedClusterMessageSender(MessageSender... messageSenders); }
LoadBalancedClusterMessageSender implements MessageSender { @Override public void onConnected() { for (MessageSender sender : senders.keySet()) { try { sender.onConnected(); } catch (Exception e) { LOG.error("Failed connecting to alpha at {}", sender.target(), e); } } } 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 void onConnected() { for (MessageSender sender : senders.keySet()) { try { sender.onConnected(); } catch (Exception e) { LOG.error("Failed connecting to alpha at {}", sender.target(), e); } } } 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);  }