### Question:
RetryableMessageSender implements MessageSender { @Override public AlphaResponse send(TxEvent event) { if (event.type() == SagaStartedEvent) { throw new OmegaException("Failed to process subsequent requests because no alpha server is available"); } try { return availableMessageSenders.take().send(event); } catch (InterruptedException e) { throw new OmegaException("Failed to send event " + event + " due to interruption", e); } catch (Exception e) { throw e; } } RetryableMessageSender(BlockingQueue<MessageSender> availableMessageSenders); @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 sendEventWhenSenderIsAvailable() { MessageSender sender = mock(MessageSender.class); availableMessageSenders.add(sender); messageSender.send(event); verify(sender, times(1)).send(event); }
@Test public void blowsUpWhenEventIsSagaStarted() { TxEvent event = new SagaStartedEvent(globalTxId, localTxId, 0); try { messageSender.send(event); expectFailing(OmegaException.class); } catch (OmegaException e) { assertThat(e.getMessage(), is("Failed to process subsequent requests because no alpha server is available")); } }
@Test public void blowsUpWhenInterrupted() 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(); }