TcpTransport implements UnpartitionedTransport, Closeable { @Override public void sendBatch(TransportBuffer buffer) throws TransportException { Buffer internalBuffer = ((TcpTransportBuffer) buffer).getInternalBuffer(); Callable<Boolean> write = () -> { try { sink.write(internalBuffer, internalBuffer.size()); return true; } catch (IOException ex) { throw new TransportException("Error while sending in tcp transport", ex); } }; try { new CallExecutor(retryConfig).execute(write); } catch (RetriesExhaustedException | UnexpectedException ue) { throw new TransportException(ue); } } TcpTransport(Sink sink, int retryCount, long retryDelayMs); @Override void sendBatch(TransportBuffer buffer); @Override void close();  }
@Test public void shouldTryAtLeastOnce() throws TransportException, IOException { Sink sink = mock(Sink.class); TcpTransport transport = new TcpTransport(sink, 0, 0); TcpTransportBuffer transportBuffer = mock(TcpTransportBuffer.class); Buffer buffer = new Buffer(); when(transportBuffer.getInternalBuffer()).thenReturn(buffer); transport.sendBatch(transportBuffer); verify(sink).write(eq(buffer), eq(0L)); }
@Test public void shouldRetry() throws TransportException, IOException { Sink sink = mock(Sink.class); TcpTransport transport = new TcpTransport(sink, 1, 0); TcpTransportBuffer transportBuffer = mock(TcpTransportBuffer.class); Buffer buffer = new Buffer(); when(transportBuffer.getInternalBuffer()).thenReturn(buffer); doThrow(new IOException()).doNothing().when(sink).write(eq(buffer), eq(0L)); transport.sendBatch(transportBuffer); verify(sink, times(2)).write(eq(buffer), eq(0L)); }
@Test public void shouldExhaustRetries() throws TransportException, IOException { Sink sink = mock(Sink.class); TcpTransport transport = new TcpTransport(sink, 4, 0); TcpTransportBuffer transportBuffer = mock(TcpTransportBuffer.class); Buffer buffer = new Buffer(); when(transportBuffer.getInternalBuffer()).thenReturn(buffer); doThrow(new IOException()).when(sink).write(eq(buffer), eq(0L)); try { transport.sendBatch(transportBuffer); fail("Should exhaust retries"); } catch (TransportException ex) { assertTrue(ex.getCause() instanceof RetriesExhaustedException); } verify(sink, times(5)).write(eq(buffer), eq(0L)); }