@Test public void processRecordsWithABadCheckpointTest() throws InterruptedException, ExecutionException { when(messageWriter.writeProcessRecordsMessage(any(ProcessRecordsInput.class))).thenReturn(buildFuture(true)); when(messageWriter.writeCheckpointMessageWithError(anyString(), anyLong(), any(Throwable.class))).thenReturn(buildFuture(false)); when(messageReader.getNextMessageFromSTDOUT()).thenAnswer(buildMessageAnswers(new ArrayList<Message>() { { this.add(new CheckpointMessage("456", 0L, null)); this.add(new StatusMessage("processRecords")); } })); assertThat(protocol.processRecords(ProcessRecordsInput.builder().records(EMPTY_RECORD_LIST) .checkpointer(checkpointer).build()), equalTo(false)); }
boolean processRecords(ProcessRecordsInput processRecordsInput) { Future<Boolean> writeFuture = messageWriter.writeProcessRecordsMessage(processRecordsInput); return waitForStatusMessage(ProcessRecordsMessage.ACTION, processRecordsInput.checkpointer(), writeFuture); }
MultiLangProtocol { boolean processRecords(ProcessRecordsInput processRecordsInput) { Future<Boolean> writeFuture = messageWriter.writeProcessRecordsMessage(processRecordsInput); return waitForStatusMessage(ProcessRecordsMessage.ACTION, processRecordsInput.checkpointer(), writeFuture); } }
MultiLangProtocol { boolean processRecords(ProcessRecordsInput processRecordsInput) { Future<Boolean> writeFuture = messageWriter.writeProcessRecordsMessage(processRecordsInput); return waitForStatusMessage(ProcessRecordsMessage.ACTION, processRecordsInput.checkpointer(), writeFuture); } MultiLangProtocol(MessageReader messageReader, MessageWriter messageWriter,
            InitializationInput initializationInput, MultiLangDaemonConfiguration configuration); }
MultiLangProtocol { boolean processRecords(ProcessRecordsInput processRecordsInput) { Future<Boolean> writeFuture = messageWriter.writeProcessRecordsMessage(processRecordsInput); return waitForStatusMessage(ProcessRecordsMessage.ACTION, processRecordsInput.checkpointer(), writeFuture); } MultiLangProtocol(MessageReader messageReader, MessageWriter messageWriter,
            InitializationInput initializationInput, MultiLangDaemonConfiguration configuration);  }
MultiLangProtocol { boolean processRecords(ProcessRecordsInput processRecordsInput) { Future<Boolean> writeFuture = messageWriter.writeProcessRecordsMessage(processRecordsInput); return waitForStatusMessage(ProcessRecordsMessage.ACTION, processRecordsInput.checkpointer(), writeFuture); } MultiLangProtocol(MessageReader messageReader, MessageWriter messageWriter,
            InitializationInput initializationInput, MultiLangDaemonConfiguration configuration);   }