Skip to content

Commit a1ca194

Browse files
garyrussellgaryrussell
committed
Fix Sonar Issues
1 parent f96edc0 commit a1ca194

File tree

6 files changed

+52
-39
lines changed

6 files changed

+52
-39
lines changed

‎spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -150,7 +150,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean {
150150
IllegalStateException isx = new IllegalStateException("Failed to determine TestUtils.boundPort() method; client version: "
151151
+ AppInfoParser.getVersion(), e);
152152
isx.addSuppressed(e1);
153-
throw isx;
153+
throw isx; // NOSONAR
154154
}
155155
}
156156
BOUND_PORT_METHOD = method;

‎spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,8 @@
144144
public class KafkaListenerAnnotationBeanPostProcessor<K, V>
145145
implements BeanPostProcessor, Ordered, ApplicationContextAware, InitializingBean, SmartInitializingSingleton {
146146

147+
private static final String UNCHECKED = "unchecked";
148+
147149
private static final String THE_LEFT = "The [";
148150

149151
private static final String RESOLVED_TO_LEFT = "Resolved to [";
@@ -643,7 +645,7 @@ private void resolveContentTypeConverter(MethodKafkaListenerEndpoint<?, ?> endpo
643645
}
644646
}
645647

646-
@SuppressWarnings({ "rawtypes", "unchecked" })
648+
@SuppressWarnings({ "rawtypes", UNCHECKED })
647649
private void resolveFilter(MethodKafkaListenerEndpoint<?, ?> endpoint, KafkaListener kafkaListener) {
648650
Object filter = resolveExpression(kafkaListener.filter());
649651
if (filter instanceof RecordFilterStrategy) {
@@ -701,7 +703,7 @@ protected String noBeanFoundMessage(Object target, String listenerBeanName, Stri
701703
+ requestedBeanName + "' was found in the application context";
702704
}
703705

704-
@SuppressWarnings("unchecked")
706+
@SuppressWarnings(UNCHECKED)
705707
private void resolveKafkaProperties(MethodKafkaListenerEndpoint<?, ?> endpoint, String[] propertyStrings) {
706708
if (propertyStrings.length > 0) {
707709
Properties properties = new Properties();
@@ -870,7 +872,7 @@ else if (relativeToCurrentValue instanceof Boolean) {
870872
return relativeToCurrent;
871873
}
872874

873-
@SuppressWarnings("unchecked")
875+
@SuppressWarnings(UNCHECKED)
874876
private void resolveAsString(Object resolvedValue, List<String> result) {
875877
if (resolvedValue instanceof String[]) {
876878
for (Object object : (String[]) resolvedValue) {
@@ -891,7 +893,7 @@ else if (resolvedValue instanceof Iterable) {
891893
}
892894
}
893895

894-
@SuppressWarnings("unchecked")
896+
@SuppressWarnings(UNCHECKED)
895897
private void resolvePartitionAsInteger(String topic, Object resolvedValue,
896898
List<TopicPartitionOffset> result, @Nullable Long offset, boolean isRelative, boolean checkDups) {
897899

‎spring-kafka/src/main/java/org/springframework/kafka/listener/DeadLetterPublishingRecoverer.java

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,8 @@
6666
*/
6767
public class DeadLetterPublishingRecoverer extends ExceptionClassifier implements ConsumerAwareRecordRecoverer {
6868

69+
private static final String DEPRECATION = "deprecation";
70+
6971
protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass())); // NOSONAR
7072

7173
private static final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition>
@@ -339,7 +341,7 @@ public void setSkipSameTopicFatalExceptions(boolean skipSameTopicFatalExceptions
339341
this.skipSameTopicFatalExceptions = skipSameTopicFatalExceptions;
340342
}
341343

342-
@SuppressWarnings({ "unchecked", "deprecation" })
344+
@SuppressWarnings({ "unchecked", DEPRECATION })
343345
@Override
344346
public void accept(ConsumerRecord<?, ?> record, @Nullable Consumer<?, ?> consumer, Exception exception) {
345347
TopicPartition tp = this.destinationResolver.apply(record, exception);
@@ -406,7 +408,7 @@ private void sendOrThrow(ProducerRecord<Object, Object> outRecord,
406408
}
407409

408410
private void maybeThrow(ConsumerRecord<?, ?> record, Exception exception) {
409-
@SuppressWarnings("deprecation")
411+
@SuppressWarnings(DEPRECATION)
410412
String message = String.format("No destination returned for record %s and exception %s. " +
411413
"failIfNoDestinationReturned: %s", ListenerUtils.recordToString(record), exception,
412414
this.throwIfNoDestinationReturned);
@@ -519,7 +521,7 @@ protected ProducerRecord<Object, Object> createProducerRecord(ConsumerRecord<?,
519521
* @param inRecord the consumer record.
520522
* @since 2.2.5
521523
*/
522-
@SuppressWarnings("deprecation")
524+
@SuppressWarnings(DEPRECATION)
523525
protected void publish(ProducerRecord<Object, Object> outRecord, KafkaOperations<Object, Object> kafkaTemplate,
524526
ConsumerRecord<?, ?> inRecord) {
525527

@@ -561,7 +563,7 @@ private void verifySendResult(KafkaOperations<Object, Object> kafkaTemplate,
561563
}
562564
}
563565

564-
@SuppressWarnings("deprecation")
566+
@SuppressWarnings(DEPRECATION)
565567
private String pubFailMessage(ProducerRecord<Object, Object> outRecord, ConsumerRecord<?, ?> inRecord) {
566568
return "Dead-letter publication to "
567569
+ outRecord.topic() + "failed for: " + ListenerUtils.recordToString(inRecord, true);

‎spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

Lines changed: 37 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,12 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
152152

153153
private static final String UNUSED = "unused";
154154

155+
private static final String DEPRECATION = "deprecation";
156+
157+
private static final String UNCHECKED = "unchecked";
158+
159+
private static final String RAWTYPES = "rawtypes";
160+
155161
private static final int DEFAULT_ACK_TIME = 5000;
156162

157163
private static final Map<String, Object> CONSUMER_CONFIG_DEFAULTS = ConsumerConfig.configDef().defaultValues();
@@ -547,12 +553,6 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
547553

548554
private static final String ERROR_HANDLER_THREW_AN_EXCEPTION = "Error handler threw an exception";
549555

550-
private static final String UNCHECKED = "unchecked";
551-
552-
private static final String RAWTYPES = "rawtypes";
553-
554-
private static final String RAW_TYPES = RAWTYPES;
555-
556556
private final LogAccessor logger = KafkaMessageListenerContainer.this.logger; // NOSONAR hide
557557

558558
private final ContainerProperties containerProperties = getContainerProperties();
@@ -611,7 +611,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
611611

612612
private final PlatformTransactionManager transactionManager = this.containerProperties.getTransactionManager();
613613

614-
@SuppressWarnings(RAW_TYPES)
614+
@SuppressWarnings(RAWTYPES)
615615
private final KafkaAwareTransactionManager kafkaTxManager =
616616
this.transactionManager instanceof KafkaAwareTransactionManager
617617
? ((KafkaAwareTransactionManager) this.transactionManager) : null;
@@ -858,7 +858,7 @@ else if (this.commonRecordInterceptor != null) {
858858
@Nullable
859859
private CommonErrorHandler determineCommonErrorHandler() {
860860
CommonErrorHandler common = getCommonErrorHandler();
861-
@SuppressWarnings("deprecation")
861+
@SuppressWarnings(DEPRECATION)
862862
GenericErrorHandler<?> errHandler = getGenericErrorHandler();
863863
if (common != null) {
864864
if (errHandler != null) {
@@ -935,7 +935,7 @@ private void checkGroupInstance(Properties properties, ConsumerFactory<K, V> con
935935
}
936936
}
937937

938-
@SuppressWarnings("deprecation")
938+
@SuppressWarnings(DEPRECATION)
939939
private boolean setupSubBatchPerPartition() {
940940
Boolean subBatching = this.containerProperties.getSubBatchPerPartition();
941941
if (subBatching != null) {
@@ -1229,7 +1229,7 @@ public boolean isLongLived() {
12291229
return true;
12301230
}
12311231

1232-
@SuppressWarnings("deprecation")
1232+
@SuppressWarnings(DEPRECATION)
12331233
@Override // NOSONAR complexity
12341234
public void run() {
12351235
ListenerUtils.setLogOnlyMetadata(this.containerProperties.isOnlyLogRecordMetadata());
@@ -1671,16 +1671,20 @@ private void pausePartitionsIfNecessary() {
16711671
}
16721672

16731673
private void resumePartitionsIfNecessary() {
1674-
List<TopicPartition> partitionsToResume = getAssignedPartitions()
1675-
.stream()
1676-
.filter(tp -> !isPartitionPauseRequested(tp)
1677-
&& this.pausedPartitions.contains(tp))
1678-
.collect(Collectors.toList());
1679-
if (partitionsToResume.size() > 0) {
1680-
this.consumer.resume(partitionsToResume);
1681-
this.pausedPartitions.removeAll(partitionsToResume);
1682-
this.logger.debug(() -> "Resumed consumption from " + partitionsToResume);
1683-
partitionsToResume.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionResumedEvent);
1674+
Collection<TopicPartition> assigned = getAssignedPartitions();
1675+
if (assigned != null) {
1676+
List<TopicPartition> partitionsToResume = assigned
1677+
.stream()
1678+
.filter(tp -> !isPartitionPauseRequested(tp)
1679+
&& this.pausedPartitions.contains(tp))
1680+
.collect(Collectors.toList());
1681+
if (partitionsToResume.size() > 0) {
1682+
this.consumer.resume(partitionsToResume);
1683+
this.pausedPartitions.removeAll(partitionsToResume);
1684+
this.logger.debug(() -> "Resumed consumption from " + partitionsToResume);
1685+
partitionsToResume
1686+
.forEach(KafkaMessageListenerContainer.this::publishConsumerPartitionResumedEvent);
1687+
}
16841688
}
16851689
}
16861690

@@ -1812,7 +1816,7 @@ record = this.acks.poll();
18121816
}
18131817
}
18141818

1815-
@SuppressWarnings("deprecation")
1819+
@SuppressWarnings(DEPRECATION)
18161820
private void traceAck(ConsumerRecord<K, V> record) {
18171821
this.logger.trace(() -> "Ack: " + ListenerUtils.recordToString(record, true));
18181822
}
@@ -1887,7 +1891,7 @@ private void processAcks(ConsumerRecords<K, V> records) {
18871891
}
18881892
}
18891893

1890-
@SuppressWarnings("deprecation")
1894+
@SuppressWarnings(DEPRECATION)
18911895
private synchronized void ackInOrder(ConsumerRecord<K, V> record) {
18921896
TopicPartition part = new TopicPartition(record.topic(), record.partition());
18931897
List<Long> offs = this.offsetsInThisBatch.get(part);
@@ -1995,7 +1999,7 @@ private void invokeBatchListener(final ConsumerRecords<K, V> recordsArg) {
19951999
}
19962000
}
19972001

1998-
@SuppressWarnings(RAW_TYPES)
2002+
@SuppressWarnings(RAWTYPES)
19992003
private void invokeBatchListenerInTx(final ConsumerRecords<K, V> records,
20002004
@Nullable final List<ConsumerRecord<K, V>> recordList) {
20012005

@@ -2306,7 +2310,7 @@ private void invokeRecordListener(final ConsumerRecords<K, V> records) {
23062310
* Invoke the listener with each record in a separate transaction.
23072311
* @param records the records.
23082312
*/
2309-
@SuppressWarnings("deprecation") // NOSONAR complexity
2313+
@SuppressWarnings(DEPRECATION) // NOSONAR complexity
23102314
private void invokeRecordListenerInTx(final ConsumerRecords<K, V> records) {
23112315
Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
23122316
while (iterator.hasNext()) {
@@ -2408,7 +2412,7 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
24082412
}
24092413
}
24102414

2411-
@SuppressWarnings("deprecation")
2415+
@SuppressWarnings(DEPRECATION)
24122416
private void doInvokeWithRecords(final ConsumerRecords<K, V> records) {
24132417
Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
24142418
while (iterator.hasNext()) {
@@ -2444,7 +2448,7 @@ private ConsumerRecords<K, V> checkEarlyIntercept(ConsumerRecords<K, V> nextArg)
24442448
return next;
24452449
}
24462450

2447-
@SuppressWarnings("deprecation")
2451+
@SuppressWarnings(DEPRECATION)
24482452
@Nullable
24492453
private ConsumerRecord<K, V> checkEarlyIntercept(ConsumerRecord<K, V> recordArg) {
24502454
internalHeaders(recordArg);
@@ -2494,7 +2498,10 @@ private void pauseForNackSleep() {
24942498
this.nackWake = System.currentTimeMillis() + this.nackSleep;
24952499
this.nackSleep = -1;
24962500
Set<TopicPartition> alreadyPaused = this.consumer.paused();
2497-
this.pausedForNack.addAll(getAssignedPartitions());
2501+
Collection<TopicPartition> assigned = getAssignedPartitions();
2502+
if (assigned != null) {
2503+
this.pausedForNack.addAll(assigned);
2504+
}
24982505
this.pausedForNack.removeAll(alreadyPaused);
24992506
this.logger.debug(() -> "Pausing for nack sleep: " + ListenerConsumer.this.pausedForNack);
25002507
try {
@@ -2605,7 +2612,7 @@ private void invokeOnMessage(final ConsumerRecord<K, V> record) {
26052612
}
26062613
}
26072614

2608-
@SuppressWarnings("deprecation")
2615+
@SuppressWarnings(DEPRECATION)
26092616
private void doInvokeOnMessage(final ConsumerRecord<K, V> recordArg) {
26102617
ConsumerRecord<K, V> record = recordArg;
26112618
if (this.recordInterceptor != null) {
@@ -2759,7 +2766,7 @@ private void sendOffsetsToTransaction() {
27592766
doSendOffsets(this.producer, commits);
27602767
}
27612768

2762-
@SuppressWarnings("deprecation")
2769+
@SuppressWarnings(DEPRECATION)
27632770
private void doSendOffsets(Producer<?, ?> prod, Map<TopicPartition, OffsetAndMetadata> commits) {
27642771
if (this.eosMode.getMode().equals(EOSMode.V1)) {
27652772
prod.sendOffsetsToTransaction(commits, this.consumerGroupId);
@@ -3171,7 +3178,7 @@ public void nack(long sleep) {
31713178
}
31723179

31733180
@Override
3174-
@SuppressWarnings("deprecation")
3181+
@SuppressWarnings(DEPRECATION)
31753182
public String toString() {
31763183
return "Acknowledgment for " + ListenerUtils.recordToString(this.record, true);
31773184
}

‎spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurerTests.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -358,6 +358,7 @@ void shouldInstantiateIfNotInContainer() {
358358
}
359359

360360
@Test
361+
@SuppressWarnings("deprecation")
361362
void shouldLogConsumerRecordMessage() {
362363
ListenerUtils.setLogOnlyMetadata(false);
363364
RetryTopicConfigurer.LoggingDltListenerHandlerMethod method =

‎spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerHolderTests.java

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,7 @@ void primary() {
9090
AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(Config3.class);
9191
MicrometerHolder micrometerHolder = new MicrometerHolder(ctx, "holderName",
9292
"timerName", "timerDesc", Collections.emptyMap());
93+
@SuppressWarnings("unchecked")
9394
Map<String, Timer> meters = (Map<String, Timer>) ReflectionTestUtils.getField(micrometerHolder, "meters");
9495
assertThat(meters).hasSize(1);
9596
}

0 commit comments

Comments
 (0)