### Question:
AbstractKinesisRecorder { public synchronized void submitAllRecords() { final RecordSender sender = getRecordSender(); final RecordIterator iterator = recordStore.iterator(); final List<byte[]> data = new ArrayList<byte[]>(MAX_RECORDS_PER_BATCH); int retry = 0; int count = 0; boolean unknownErrorRetried = false; try { while (iterator.hasNext() && retry < MAX_RETRY_COUNT) { final String streamName = nextBatch(iterator, data, MAX_RECORDS_PER_BATCH, MAX_BATCH_RECORDS_SIZE_BYTES); if (streamName == null || data.isEmpty()) { break; } try { List<byte[]> failures = null; try { failures = sender.sendBatch(streamName, data); } catch (AmazonClientException ace) { if (!unknownErrorRetried && ace.getMessage() != null && ace.getMessage().contains("Unable to unmarshall error response")) { unknownErrorRetried = true; failures = data; } else { throw ace; } } final int successCount = data.size() - failures.size(); count += successCount; iterator.removeReadRecords(); if (successCount == 0) { retry++; } if (!failures.isEmpty()) { for (final byte[] bytes : failures) { saveRecord(bytes, streamName); } } } catch (final AmazonClientException ace) { if (sender.isRecoverable(ace)) { LOGGER.error( "ServiceException in submit all, the values of the data inside the requests appears valid. The request will be kept", ace); } else { try { this.config.getDeadLetterListener().onRecordsDropped(streamName, data); } catch (Exception e) { LOGGER.error("DeadLetterListener onRecordsDropped has thrown an exception (user code)", e); } try { iterator.removeReadRecords(); } catch (final IOException e) { throw new AmazonClientException("Failed to drop bad records.", e); } LOGGER.error( "ServiceException in submit all, the last request is presumed to be the cause and will be dropped", ace); } throw ace; } catch (final IOException e) { throw new AmazonClientException("Failed to remove read records", e); } } } finally { LOGGER.debug(String.format("submitAllRecords sent %d records", count)); try { iterator.close(); } catch (final IOException e) { throw new AmazonClientException("Failed to close record file", e); } } } protected  AbstractKinesisRecorder(FileRecordStore recordStore, KinesisRecorderConfig config); void saveRecord(String data, String streamName); void saveRecord(byte[] data, String streamName); synchronized void submitAllRecords(); KinesisRecorderConfig getKinesisRecorderConfig(); long getDiskBytesUsed(); long getDiskByteLimit(); synchronized void deleteAllRecords();  }### Answer:
@Test public void testSubmitAllRecords() { for (int i = 0; i < 10; i++) { recorder.saveRecord(randomBytes(1024), STREAM_NAME); } String anotherStream = "another_stream"; for (int i = 0; i < 10; i++) { recorder.saveRecord(randomBytes(1024), anotherStream); } Mockito.when(sender.sendBatch(Mockito.anyString(), Mockito.anyListOf(byte[].class))) .thenReturn(new ArrayList<byte[]>()); recorder.submitAllRecords(); assertEquals("no records after submitAllRecords", 0, recorder.getDiskBytesUsed()); }