### Question:
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);   }### Answer:
@Test public void processRecordsTest() throws InterruptedException, ExecutionException { when(messageWriter.writeProcessRecordsMessage(any(ProcessRecordsInput.class))).thenReturn(buildFuture(true)); when(messageReader.getNextMessageFromSTDOUT()).thenReturn(buildFuture( new StatusMessage("processRecords"), Message.class)); assertThat(protocol.processRecords(ProcessRecordsInput.builder().records(EMPTY_RECORD_LIST).build()), equalTo(true)); }
@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)); }
@Test(expected = NullPointerException.class) public void waitForStatusMessageTimeoutTest() throws InterruptedException, TimeoutException, ExecutionException { when(messageWriter.writeProcessRecordsMessage(any(ProcessRecordsInput.class))).thenReturn(buildFuture(true)); Future<Message> future = Mockito.mock(Future.class); when(messageReader.getNextMessageFromSTDOUT()).thenReturn(future); when(configuration.getTimeoutInSeconds()).thenReturn(5); when(future.get(anyInt(), eq(TimeUnit.SECONDS))).thenThrow(TimeoutException.class); protocol = new MultiLangProtocolForTesting(messageReader, messageWriter, InitializationInput.builder().shardId(shardId).build(), configuration); protocol.processRecords(ProcessRecordsInput.builder().records(EMPTY_RECORD_LIST).build()); }
@Test public void waitForStatusMessageSuccessTest() { when(messageWriter.writeProcessRecordsMessage(any(ProcessRecordsInput.class))).thenReturn(buildFuture(true)); when(messageReader.getNextMessageFromSTDOUT()).thenReturn(buildFuture( new StatusMessage("processRecords"), Message.class)); when(configuration.getTimeoutInSeconds()).thenReturn(5); assertTrue(protocol.processRecords(ProcessRecordsInput.builder().records(EMPTY_RECORD_LIST).build())); }