Cleanup Kinesis Logs
@@ -159,23 +159,23 @@ public class KinesisConsumer extends KinesisConnectionHandler implements Partiti
return new KinesisRecordsBatch(recordList, startShardToSequenceNum.getKey(), isEndOfShard);
} catch (IllegalStateException e) {
- LOGGER.warn("Illegal state exception, connection is broken", e);
+ debugOrLogWarning("Illegal state exception, connection is broken", e);
return handleException(kinesisStartCheckpoint, recordList);
} catch (ProvisionedThroughputExceededException e) {
- LOGGER.warn("The request rate for the stream is too high", e);
+ debugOrLogWarning("The request rate for the stream is too high", e);
return handleException(kinesisStartCheckpoint, recordList);
} catch (ExpiredIteratorException e) {
- LOGGER.warn("ShardIterator expired while trying to fetch records", e);
+ debugOrLogWarning("ShardIterator expired while trying to fetch records", e);
return handleException(kinesisStartCheckpoint, recordList);
} catch (ResourceNotFoundException | InvalidArgumentException e) {
// aws errors
LOGGER.error("Encountered AWS error while attempting to fetch records", e);
return handleException(kinesisStartCheckpoint, recordList);
} catch (KinesisException e) {
- LOGGER.warn("Encountered unknown unrecoverable AWS exception", e);
+ debugOrLogWarning("Encountered unknown unrecoverable AWS exception", e);
throw new RuntimeException(e);
} catch (AbortedException e) {
- LOGGER.warn("Task aborted due to exception.", e);
+ debugOrLogWarning("Task aborted due to exception", e);
return handleException(kinesisStartCheckpoint, recordList);
} catch (Throwable e) {
// non transient errors
@@ -184,6 +184,14 @@ public class KinesisConsumer extends KinesisConnectionHandler implements Partiti
}
}
+ private void debugOrLogWarning(String message, Throwable throwable) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(message, throwable);
+ } else {
+ LOGGER.warn(message + ": " + throwable.getMessage());
+ }
+ }
+
private KinesisRecordsBatch handleException(KinesisPartitionGroupOffset start, List<Record> recordList) {
String shardId = start.getShardToStartSequenceMap().entrySet().iterator().next().getKey();
