diff --git a/build.gradle b/build.gradle index 1cd8419..e5403fe 100644 --- a/build.gradle +++ b/build.gradle @@ -3,11 +3,11 @@ plugins { id 'eclipse' id 'idea' id 'jacoco' - id 'org.sonarqube' version '3.0' + id 'org.sonarqube' version '3.1.1' id 'checkstyle' id 'org.ajoberstar.grgit' version '4.1.0' id "io.spring.dependency-management" version '1.0.11.RELEASE' - id 'com.jfrog.artifactory' version '4.18.2' + id 'com.jfrog.artifactory' version '4.19.0' } description = 'Spring Integration AWS Support' @@ -27,15 +27,15 @@ ext { assertjVersion = '3.18.1' awaitilityVersion = '4.0.3' dynamodbLockClientVersion = '1.1.0' - jacksonVersion = '2.11.3' - junitVersion = '5.6.2' + jacksonVersion = '2.12.1' + junitVersion = '5.7.0' servletApiVersion = '4.0.1' - localstackVersion = '0.2.5' - log4jVersion = '2.13.3' + localstackVersion = '0.2.6' + log4jVersion = '2.14.0' springCloudAwsVersion = '2.2.5.RELEASE' - springIntegrationVersion = '5.3.5.RELEASE' + springIntegrationVersion = '5.4.3' kinesisClientVersion = '1.14.0' - kinesisProducerVersion = '0.14.1' + kinesisProducerVersion = '0.14.3' idPrefix = 'aws' @@ -96,7 +96,7 @@ jacoco { checkstyle { configDirectory.set(rootProject.file("src/checkstyle")) - toolVersion = '8.38' + toolVersion = '8.39' } dependencies { diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 4d9ca16..28ff446 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,5 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-6.7.1-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-6.8.1-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java index fb243ff..1aab4ac 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -86,7 +86,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { private static final ThreadLocal attributesHolder = new ThreadLocal<>(); /** - * Interval to run lease cleanup thread in {@link LeaseCleanupManager}. + * Interval to run lease cleanup thread in {@link com.amazonaws.services.kinesis.leases.impl.LeaseCleanupManager}. */ private static final long DEFAULT_LEASE_CLEANUP_INTERVAL_MILLIS = Duration.ofMinutes(1).toMillis(); @@ -198,7 +198,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { public void setConsumerGroup(String consumerGroup) { Assert.hasText(consumerGroup, "'consumerGroup' must not be empty"); Assert.isNull(this.config, "'consumerGroup' must be configured as an application name " + - "on the provided KinesisClientLibConfiguration"); + "on the provided KinesisClientLibConfiguration"); this.consumerGroup = consumerGroup; } @@ -405,16 +405,12 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { @Override public void initialize(String shardId) { this.shardId = shardId; - if (logger.isInfoEnabled()) { - logger.info("Initializing record processor for shard: " + this.shardId); - } + logger.info(() -> "Initializing record processor for shard: " + this.shardId); } @Override public void processRecords(List records, IRecordProcessorCheckpointer checkpointer) { - if (logger.isDebugEnabled()) { - logger.debug("Processing " + records.size() + " records from " + this.shardId); - } + logger.debug(() -> "Processing " + records.size() + " records from " + this.shardId); try { if (ListenerMode.record.equals(KclMessageDrivenChannelAdapter.this.listenerMode)) { @@ -485,8 +481,8 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { } payload = messageToUse.getPayload(); } - catch (Exception e) { - logger.warn("Could not parse embedded headers. Remain payload untouched.", e); + catch (Exception ex) { + logger.warn(ex, "Could not parse embedded headers. Remain payload untouched."); } } @@ -523,9 +519,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { try { sendMessage(messageToSend); } - catch (Exception e) { - logger.error("Got an exception during sending a '" + messageToSend + "'" + "\nfor the '" + rawRecord - + "'.\n" + "Consider to use 'errorChannel' flow for the compensation logic.", e); + catch (Exception ex) { + logger.error(ex, () -> "Got an exception during sending a '" + messageToSend + "'" + "\nfor the '" + + rawRecord + "'.\n" + "Consider to use 'errorChannel' flow for the compensation logic."); } } @@ -549,9 +545,7 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { * @param record last processed record */ private void checkpoint(IRecordProcessorCheckpointer checkpointer, @Nullable Record record) { - if (logger.isInfoEnabled()) { - logger.info("Checkpointing shard " + this.shardId); - } + logger.info(() -> "Checkpointing shard " + this.shardId); try { if (record == null) { checkpointer.checkpoint(); @@ -563,19 +557,14 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { catch (ShutdownException se) { // Ignore checkpoint if the processor instance has been shutdown (fail // over). - logger.info("Caught shutdown exception, skipping checkpoint.", se); + logger.info(se, "Caught shutdown exception, skipping checkpoint."); } - catch (ThrottlingException e) { - if (logger.isInfoEnabled()) { - logger.info("Transient issue when checkpointing", e); - } + catch (ThrottlingException ex) { + logger.info(ex, "Transient issue when checkpointing"); } - catch (InvalidStateException e) { - // This indicates an issue with the DynamoDB table (check for table, - // provisioned - // IOPS). - logger.error("Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library.", - e); + catch (InvalidStateException ex) { + // This indicates an issue with the DynamoDB table (check for table, provisioned IOPS). + logger.error(ex, "Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client."); } } @@ -602,14 +591,12 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport { @Override public void shutdown(IRecordProcessorCheckpointer checkpointer, ShutdownReason reason) { - if (logger.isInfoEnabled()) { - logger.info("Scheduler is shutting down for reason '" + reason + "'; checkpointing..."); - } + logger.info(() -> "Scheduler is shutting down for reason '" + reason + "'; checkpointing..."); try { checkpointer.checkpoint(); } - catch (ShutdownException | InvalidStateException e) { - logger.error("Exception while checkpointing at requested shutdown. Giving up", e); + catch (ShutdownException | InvalidStateException ex) { + logger.error(ex, "Exception while checkpointing at requested shutdown. Giving up"); } } diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index 530cdbd..63dd706 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -368,17 +368,18 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport protected void onInit() { super.onInit(); + final String componentName = getComponentName(); if (this.consumerExecutor == null) { this.consumerExecutor = Executors.newCachedThreadPool( new CustomizableThreadFactory( - (getComponentName() == null ? "" : getComponentName()) + "-kinesis-consumer-")); + (componentName == null ? "" : componentName) + "-kinesis-consumer-")); } if (this.dispatcherExecutor == null) { this.dispatcherExecutor = Executors.newCachedThreadPool( new CustomizableThreadFactory( - (getComponentName() == null ? "" : getComponentName()) + "-kinesis-dispatcher-")); + (componentName == null ? "" : componentName) + "-kinesis-dispatcher-")); } if (this.streams == null) { @@ -407,14 +408,12 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport shardConsumer.stop(); } else { - if (this.logger.isDebugEnabled()) { - this.logger.debug( + this.logger.debug(() -> "There is no ShardConsumer for shard [" + shard + "] in stream [" + shard + "] to stop."); - } } } @@ -423,9 +422,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport KinesisShardOffset shardOffsetForSearch = KinesisShardOffset.latest(stream, shard); ShardConsumer shardConsumer = this.shardConsumers.get(shardOffsetForSearch); if (shardConsumer != null) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("The [" + shardConsumer + "] has been started before."); - } + this.logger.debug(() -> "The [" + shardConsumer + "] has been started before."); } else { synchronized (this.shardOffsets) { @@ -473,9 +470,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport + shardOffset.getStream() + "]"); - if (logger.isDebugEnabled()) { - logger.debug("Resetting consumer for [" + shardOffset + "]..."); - } + logger.debug(() -> "Resetting consumer for [" + shardOffset + "]..."); shardOffset.reset(); synchronized (this.shardOffsets) { this.shardOffsets.remove(shardOffset); @@ -580,8 +575,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } catch (LimitExceededException limitExceededException) { - - logger.info( + logger.info(() -> "Got LimitExceededException when listing stream [" + stream + "]. " @@ -646,12 +640,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport boolean skipClosedAndExhaustedShard = checkpoint != null && new BigInteger(endingSequenceNumber) .compareTo(new BigInteger(checkpoint)) <= 0; - if (logger.isTraceEnabled()) { - logger.trace("The shard [" + shard + "] in stream [" + stream + logger.trace(() -> "The shard [" + shard + "] in stream [" + stream + "] is closed CLOSED and exhausted with endingSequenceNumber [" + endingSequenceNumber + "].\nThe last processed checkpoint is [" + checkpoint + "]." + (skipClosedAndExhaustedShard ? "\nThe shard will be skipped." : "")); - } if (skipClosedAndExhaustedShard) { // Skip CLOSED shard which has been exhausted @@ -663,11 +655,11 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport shardsToConsume.add(shard); } } - catch (Exception e) { + catch (Exception ex) { String exceptionMessage = "Got an exception when processing shards in stream [" + stream + "]"; - logger.info(exceptionMessage + ".\n Retrying... ", e); + logger.info(ex, () -> exceptionMessage + ".\n Retrying... "); if (retry > 5) { - throw new IllegalStateException(exceptionMessage, e); + throw new IllegalStateException(exceptionMessage, ex); } //Retry detectShardsToConsume(stream, retry + 1); @@ -681,16 +673,16 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { Thread.sleep(sleepAmount); } - catch (Exception e) { + catch (Exception ex) { if (interruptThread) { Thread.currentThread().interrupt(); } if (this.active) { - logger.error(error.getMessage(), e); + logger.error(ex, error.getMessage()); } else { - logger.info(error.getMessage() + " while adapter was inactive", e); + logger.info(ex, () -> error.getMessage() + " while adapter was inactive"); } throw error; @@ -716,7 +708,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } catch (Exception ex) { - logger.error("Error population shards for stream: " + stream, ex); + logger.error(ex, () -> "Error population shards for stream: " + stream); } finally { if (shardsGatherLatch != null) { @@ -852,9 +844,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport for (String stream : KinesisMessageDrivenChannelAdapter.this.inResharding) { // Local store to avoid several tasks for the same 'stream' if (this.inReshardingProcess.add(stream)) { - if (logger.isDebugEnabled()) { - logger.debug("Resharding has happened for stream [" + stream + "]. Rebalancing..."); - } + logger.debug(() -> "Resharding has happened for stream [" + stream + "]. Rebalancing..."); populateShardsForStream(stream, null); } } @@ -938,7 +928,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport if (ex instanceof InterruptedException) { Thread.currentThread().interrupt(); } - logger.info("The lock for key '" + this.key + "' was not unlocked in time", ex); + logger.info(ex, () -> "The lock for key '" + this.key + "' was not unlocked in time"); } } if (this.notifier != null) { @@ -973,8 +963,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.shardOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); } } - if (logger.isInfoEnabled() && this.state == ConsumerState.NEW) { - logger.info("The [" + this + "] has been started."); + if (this.state == ConsumerState.NEW) { + logger.info(() -> "The [" + this + "] has been started."); } GetShardIteratorRequest shardIteratorRequest = this.shardOffset.toShardIteratorRequest(); @@ -1006,19 +996,15 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport case STOP: if (this.shardIterator == null) { - if (logger.isInfoEnabled()) { - logger.info( - "Stopping the [" - + this - + "] on the checkpoint [" - + this.checkpointer.getCheckpoint() - + "] because the shard has been CLOSED and exhausted."); - } + logger.info(() -> + "Stopping the [" + + this + + "] on the checkpoint [" + + this.checkpointer.getCheckpoint() + + "] because the shard has been CLOSED and exhausted."); } else { - if (logger.isInfoEnabled()) { - logger.info("Stopping the [" + this + "]."); - } + logger.info(() -> "Stopping the [" + this + "]."); } this.task = null; break; @@ -1048,7 +1034,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport if (ex instanceof InterruptedException) { Thread.currentThread().interrupt(); } - logger.info("The lock for key '" + this.key + "' was not renewed in time", ex); + logger.info(ex, () -> "The lock for key '" + this.key + "' was not renewed in time"); } if (!lockRenewed && this.state == ConsumerState.CONSUME) { @@ -1120,16 +1106,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } if (ConsumerState.STOP != this.state && result.getRecords().isEmpty()) { - if (logger.isDebugEnabled()) { - logger.debug( - "No records for [" - + this - + "] on sequenceNumber [" - + this.checkpointer.getLastCheckpointValue() - + "]. Suspend consuming for [" - + KinesisMessageDrivenChannelAdapter.this.consumerBackoff - + "] milliseconds."); - } + logger.debug(() -> + "No records for [" + + this + + "] on sequenceNumber [" + + this.checkpointer.getLastCheckpointValue() + + "]. Suspend consuming for [" + + KinesisMessageDrivenChannelAdapter.this.consumerBackoff + + "] milliseconds."); prepareSleepState(); } } @@ -1144,11 +1128,11 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.checkpointer.checkpoint(endingSequenceNumber); } catch (ProvisionedThroughputExceededException ignored) { - // This exception is ignored to gurantee that an exhausted shard is marked as CLOSED + // This exception is ignored to guarantee that an exhausted shard is marked as CLOSED // even in the case it's not possible to checkpoint. Otherwise the ShardConsumer is // left in an illegal state where the shard iterator is null without any possibility // of recovering from it. - logger.debug("Exception while checkpointing empty shards", ignored); + logger.debug(ignored, "Exception while checkpointing empty shards"); } } @@ -1161,23 +1145,19 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport // records. // Lets acquire iterator again (using checkpointer for iterator start // sequence number). - if (logger.isInfoEnabled()) { - logger.info( + logger.info(() -> "Shard iterator for [" + ShardConsumer.this + "] expired.\n" + "A new one will be started from the check pointed sequence number."); - } this.state = ConsumerState.EXPIRED; } - catch (ProvisionedThroughputExceededException e) { - if (logger.isWarnEnabled()) { - logger.warn( + catch (ProvisionedThroughputExceededException ex) { + logger.warn(() -> "GetRecords request throttled for [" + ShardConsumer.this + "] with the reason: " - + e.getErrorMessage()); - } + + ex.getErrorMessage()); // We are throttled, so let's sleep prepareSleepState(); } @@ -1192,9 +1172,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } private void processRecords(List records) { - if (logger.isTraceEnabled()) { - logger.trace("Processing records: " + records + " for [" + ShardConsumer.this + "]"); - } + logger.trace(() -> "Processing records: " + records + " for [" + ShardConsumer.this + "]"); this.checkpointer.setHighestSequence(records.get(records.size() - 1).getSequenceNumber()); @@ -1266,8 +1244,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport payload = messageToUse.getPayload(); } - catch (Exception e) { - logger.warn("Could not parse embedded headers. Remain payload untouched.", e); + catch (Exception ex) { + logger.warn(ex, "Could not parse embedded headers. Remain payload untouched."); } } @@ -1307,16 +1285,15 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { sendMessage(messageToSend); } - catch (Exception e) { - logger.info( + catch (Exception ex) { + logger.info(ex, () -> "Got an exception during sending a '" + messageToSend + "'" + "\nfor the '" + rawRecord + "'.\n" - + "Consider to use 'errorChannel' flow for the compensation logic.", - e); + + "Consider to use 'errorChannel' flow for the compensation logic."); } } @@ -1400,10 +1377,10 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { shardConsumer.task.run(); } - catch (Exception e) { - logger.info( + catch (Exception ex) { + logger.info(() -> "Got an exception " - + e + + ex + " during [" + shardConsumer + "] task invocation" @@ -1482,8 +1459,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } } - catch (Exception e) { - logger.error("Error during locking: " + lock, e); + catch (Exception ex) { + logger.error(ex, "Error during locking: " + lock); } } @@ -1502,8 +1479,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { lock.unlock(); } - catch (Exception e) { - logger.error("Error during unlocking: " + lock, e); + catch (Exception ex) { + logger.error(ex, "Error during unlocking: " + lock); } } forUnlocking.complete(true); @@ -1532,9 +1509,9 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport this.locks.remove(lockFuture.lockKey); } } - catch (Exception e) { + catch (Exception ex) { lockFuture.complete(false); - logger.error("Error during locking: " + lock, e); + logger.error(ex, () -> "Error during locking: " + lock); } } else { @@ -1557,8 +1534,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport try { lock.unlock(); } - catch (Exception e) { - logger.error("Error during unlocking: " + lock, e); + catch (Exception ex) { + logger.error(ex, () -> "Error during unlocking: " + lock); } finally { iterator.remove();