BatchTimeoutPolicy implements FlushPolicy<MESSAGE_T> { protected long getBatchTimeout(String sensorType, WriterConfiguration configurations) { int batchTimeoutSecs = configurations.getBatchTimeout(sensorType); if (batchTimeoutSecs <= 0 || batchTimeoutSecs > maxBatchTimeout) { LOG.debug("The configured batch timeout '{}' for sensor type '{}' is <=0 or > the maximum allowable batch timeout '{}'. Setting the batch timeout to the maximum allowable.", batchTimeoutSecs, sensorType, maxBatchTimeout); batchTimeoutSecs = maxBatchTimeout; } return TimeUnit.SECONDS.toMillis(batchTimeoutSecs); } BatchTimeoutPolicy(int maxBatchTimeout);  BatchTimeoutPolicy(int maxBatchTimeout, Clock clock); @Override boolean shouldFlush(String sensorType, WriterConfiguration configurations, List<BulkMessage<MESSAGE_T>> messages); @Override void onFlush(String sensorType, BulkWriterResponse response);  }
@Test public void getBatchTimeoutShouldReturnConfiguredTimeout() { BatchTimeoutPolicy batchTimeoutPolicy = new BatchTimeoutPolicy(maxBatchTimeout); when(configurations.getBatchTimeout(sensor1)).thenReturn(5); assertEquals(5000L, batchTimeoutPolicy.getBatchTimeout(sensor1, configurations)); }
@Test public void getBatchTimeoutShouldReturnMaxBatchTimeout() { BatchTimeoutPolicy batchTimeoutPolicy = new BatchTimeoutPolicy(maxBatchTimeout); when(configurations.getBatchTimeout(sensor1)).thenReturn(0); assertEquals(maxBatchTimeout * 1000, batchTimeoutPolicy.getBatchTimeout(sensor1, configurations)); }