diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java index 31133d58d..e37b95eb7 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisioner.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2023 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,6 +86,7 @@ import org.springframework.util.StringUtils; * @author Aldo Sinanaj * @author Yi Liu * @author Omer Celik + * @author Byungjun You */ public class KafkaTopicProvisioner implements // @checkstyle:off @@ -330,8 +331,8 @@ public class KafkaTopicProvisioner implements dlqProducerProperties.getTopic()); } catch (Throwable throwable) { - if (throwable instanceof Error) { - throw (Error) throwable; + if (throwable instanceof Error throwableError) { + throw throwableError; } else { throw new ProvisioningException("Provisioning exception encountered for " + name, throwable); @@ -352,8 +353,8 @@ public class KafkaTopicProvisioner implements // TODO: // https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/pull/514#discussion_r241075940 catch (Throwable throwable) { - if (throwable instanceof Error) { - throw (Error) throwable; + if (throwable instanceof Error throwableError) { + throw throwableError; } else { // TODO: @@ -570,8 +571,8 @@ public class KafkaTopicProvisioner implements describeTopicsResult.all().get(); } catch (ExecutionException ex) { - if (ex.getCause() instanceof UnknownTopicOrPartitionException) { - throw (UnknownTopicOrPartitionException) ex.getCause(); + if (ex.getCause() instanceof UnknownTopicOrPartitionException unknownTopicOrPartitionException) { + throw unknownTopicOrPartitionException; } else { logger.warn("No partitions have been retrieved for the topic " diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 25d7f7714..e46553710 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2022 the original author or authors. + * Copyright 2021-2023 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. @@ -28,7 +28,6 @@ import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; import reactor.kafka.receiver.KafkaReceiver; import reactor.kafka.receiver.ReceiverOptions; @@ -67,6 +66,7 @@ import org.springframework.util.StringUtils; /** * @author Gary Russell + * @author Byungjun You * @since 4.0 * */ @@ -223,10 +223,7 @@ public class ReactorKafkaBinder protected void handleMessageInternal(Message message) { Object sendResultHeader = message.getHeaders().get("sendResult"); Sinks.One sink = Sinks.one(); - if (sendResultHeader instanceof AtomicReference) { - @SuppressWarnings("unchecked") - AtomicReference> result = - (AtomicReference>) sendResultHeader; + if (sendResultHeader instanceof AtomicReference result) { result.set(sink.asMono()); } if (this.sender != null) { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 85d038edf..4eb1ff562 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2023 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. @@ -81,6 +81,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Soby Chacko * @author Gary Russell + * @author Byungjun You */ @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ KafkaProperties.class, KafkaStreamsExtendedBindingProperties.class }) @@ -155,8 +156,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration { @SuppressWarnings("unchecked") private void flatten(String propertyName, Object value, Map flattenedProperties) { - if (value instanceof Map) { - ((Map) value).forEach((k, v) -> flatten( + if (value instanceof Map valueAsMap) { + valueAsMap.forEach((k, v) -> flatten( (propertyName != null ? propertyName + "." : "") + k, v, flattenedProperties)); } @@ -213,10 +214,7 @@ public class KafkaStreamsBinderSupportAutoConfiguration { else { Object bootstrapServerConfig = properties .get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); - if (bootstrapServerConfig instanceof String) { - @SuppressWarnings("unchecked") - String bootStrapServers = (String) properties - .get(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG); + if (bootstrapServerConfig instanceof String bootStrapServers) { if (bootStrapServers.equals("localhost:9092")) { properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionString); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java index b07a8449d..1511ee954 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2022 the original author or authors. + * Copyright 2019-2023 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. @@ -65,6 +65,7 @@ import org.springframework.util.CollectionUtils; /** * @author Soby Chacko + * @author Byungjun You * @since 2.2.0 */ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderProcessor implements BeanFactoryAware { @@ -397,8 +398,8 @@ public class KafkaStreamsFunctionProcessor extends AbstractKafkaStreamsBinderPro private Object handleCurriedFunctions(Object[] adaptedInboundArguments, Object result) { int i = 1; while (result instanceof Function || result instanceof Consumer) { - if (result instanceof Function) { - result = ((Function) result).apply(adaptedInboundArguments[i]); + if (result instanceof Function function) { + result = function.apply(adaptedInboundArguments[i]); } else { ((Consumer) result).accept(adaptedInboundArguments[i]); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java index d805d9c14..60cf9a5b8 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2023 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. @@ -50,6 +50,7 @@ import org.springframework.util.StringUtils; * contentType. Based on the contentType, a {@link MessageConverter} will be resolved. * * @author Soby Chacko + * @author Byungjun You */ public class KafkaStreamsMessageConversionDelegate { @@ -94,7 +95,7 @@ public class KafkaStreamsMessageConversionDelegate { final KStream kStreamWithEnrichedHeaders = outboundBindTarget .filter((k, v) -> v != null) .mapValues((v) -> { - Message message = v instanceof Message ? (Message) v + Message message = v instanceof Message m ? m : MessageBuilder.withPayload(v).build(); Map headers = new HashMap<>(message.getHeaders()); if (StringUtils.hasText(contentType)) { @@ -171,14 +172,14 @@ public class KafkaStreamsMessageConversionDelegate { if (o2 instanceof Message || o2 instanceof String || o2 instanceof byte[]) { Message m1 = null; - if (o2 instanceof Message) { + if (o2 instanceof Message message) { m1 = perRecordContentTypeHolder.contentType != null - ? MessageBuilder.fromMessage((Message) o2) + ? MessageBuilder.fromMessage(message) .setHeader( MessageHeaders.CONTENT_TYPE, perRecordContentTypeHolder.contentType) .build() - : (Message) o2; + : message; } else { m1 = perRecordContentTypeHolder.contentType != null diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/BinderHeaderMapper.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/BinderHeaderMapper.java index 9e5dd435c..99dde2d1a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/BinderHeaderMapper.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/BinderHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 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. @@ -62,6 +62,7 @@ import org.springframework.util.MimeType; * @author Gary Russell * @author Artem Bilan * @author Soby Chacko + * @author Byungjun You * * @since 3.0.0 * @@ -256,8 +257,8 @@ public class BinderHeaderMapper extends AbstractKafkaHeaderMapper { headers.forEach((key, rawValue) -> { if (matches(key, rawValue)) { Object valueToAdd = headerValueToAddOut(key, rawValue); - if (valueToAdd instanceof byte[]) { - target.add(new RecordHeader(key, (byte[]) valueToAdd)); + if (valueToAdd instanceof byte[] valueToAddBytes) { + target.add(new RecordHeader(key, valueToAddBytes)); } else { try { @@ -268,8 +269,8 @@ public class BinderHeaderMapper extends AbstractKafkaHeaderMapper { } if (!this.encodeStrings && !MimeType.class.isAssignableFrom(rawValue.getClass()) - && valueToAdd instanceof String) { - target.add(new RecordHeader(key, ((String) valueToAdd).getBytes(getCharset()))); + && valueToAdd instanceof String stringValueToAdd) { + target.add(new RecordHeader(key, stringValueToAdd.getBytes(getCharset()))); className = JAVA_LANG_STRING; } else { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 61f90a679..920a3695f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2022 the original author or authors. + * Copyright 2014-2023 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. @@ -155,6 +155,7 @@ import org.springframework.util.backoff.FixedBackOff; * @author Taras Danylchuk * @author Yi Liu * @author Chris Bono + * @author Byungjun You */ public class KafkaMessageChannelBinder extends // @checkstyle:off @@ -430,13 +431,11 @@ public class KafkaMessageChannelBinder extends List interceptors = ((InterceptableChannel) channel) .getInterceptors(); interceptors.forEach((interceptor) -> { - if (interceptor instanceof PartitioningInterceptor) { - ((PartitioningInterceptor) interceptor) - .setPartitionCount(partitions.size()); + if (interceptor instanceof PartitioningInterceptor partitioningInterceptor) { + partitioningInterceptor.setPartitionCount(partitions.size()); } - else if (interceptor instanceof DefaultPartitioningInterceptor) { - ((DefaultPartitioningInterceptor) interceptor) - .setPartitionCount(partitions.size()); + else if (interceptor instanceof DefaultPartitioningInterceptor defaultPartitioningInterceptor) { + defaultPartitioningInterceptor.setPartitionCount(partitions.size()); } }); } @@ -721,9 +720,8 @@ public class KafkaMessageChannelBinder extends if (!extendedConsumerProperties.isBatchMode() && extendedConsumerProperties.getMaxAttempts() > 1 && transMan == null) { - if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) - || ((ListenerContainerWithDlqAndRetryCustomizer) customizer) - .retryAndDlqInBinding(destination.getName(), group)) { + if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer c) + || c.retryAndDlqInBinding(destination.getName(), group)) { kafkaMessageDrivenChannelAdapter .setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); kafkaMessageDrivenChannelAdapter @@ -770,15 +768,14 @@ public class KafkaMessageChannelBinder extends CommonErrorHandler.class); messageListenerContainer.setCommonErrorHandler(commonErrorHandler); } - if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) { + if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer c) { BiFunction, Exception, TopicPartition> destinationResolver = createDestResolver( extendedConsumerProperties.getExtension()); BackOff createBackOff = extendedConsumerProperties.getMaxAttempts() > 1 ? createBackOff(extendedConsumerProperties) : null; - ((ListenerContainerWithDlqAndRetryCustomizer) customizer) - .configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver, + c.configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver, createBackOff); } else { @@ -1067,8 +1064,8 @@ public class KafkaMessageChannelBinder extends MessageConverter messageConverter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), extendedConsumerProperties, this.configurationProperties); - if (messageConverter instanceof MessagingMessageConverter) { - ((MessagingMessageConverter) messageConverter).setHeaderMapper(getHeaderMapper(extendedConsumerProperties)); + if (messageConverter instanceof MessagingMessageConverter messagingMessageConverter) { + messagingMessageConverter.setHeaderMapper(getHeaderMapper(extendedConsumerProperties)); } return messageConverter; } @@ -1139,11 +1136,11 @@ public class KafkaMessageChannelBinder extends Object timeout = producerFactory.getConfigurationProperties().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG); Long sendTimeout = null; - if (timeout instanceof Number) { - sendTimeout = ((Number) timeout).longValue() + 2000L; + if (timeout instanceof Number timeoutAsNumber) { + sendTimeout = timeoutAsNumber.longValue() + 2000L; } - else if (timeout instanceof String) { - sendTimeout = Long.parseLong((String) timeout) + 2000L; + else if (timeout instanceof String timeoutAsString) { + sendTimeout = Long.parseLong(timeoutAsString) + 2000L; } if (timeout == null) { sendTimeout = ((Integer) ProducerConfig.configDef() @@ -1197,9 +1194,9 @@ public class KafkaMessageChannelBinder extends AtomicReference> recordToSend = new AtomicReference<>( record); Throwable throwable = null; - if (message.getPayload() instanceof Throwable) { + if (message.getPayload() instanceof Throwable throwablePayload) { - throwable = (Throwable) message.getPayload(); + throwable = throwablePayload; HeaderMode headerMode = properties.getHeaderMode(); @@ -1338,11 +1335,9 @@ public class KafkaMessageChannelBinder extends } } else { - if (message.getPayload() instanceof MessagingException) { + if (message.getPayload() instanceof MessagingException messagingException) { AcknowledgmentCallback ack = StaticMessageHeaderAccessor - .getAcknowledgmentCallback( - ((MessagingException) message.getPayload()) - .getFailedMessage()); + .getAcknowledgmentCallback(messagingException.getFailedMessage()); if (ack != null) { if (isAutoCommitOnError(properties)) { ack.acknowledge(AcknowledgmentCallback.Status.REJECT); @@ -1514,9 +1509,9 @@ public class KafkaMessageChannelBinder extends @Override public void stop() { - if (this.producerFactory instanceof DisposableBean) { + if (this.producerFactory instanceof DisposableBean disposableProducerFactory) { try { - ((DisposableBean) producerFactory).destroy(); + disposableProducerFactory.destroy(); } catch (Exception ex) { this.logger.error(ex, "Error destroying the producer factory bean: "); @@ -1636,8 +1631,8 @@ public class KafkaMessageChannelBinder extends } private String keyOrValue(Object keyOrValue) { - if (keyOrValue instanceof byte[]) { - return "byte[" + ((byte[]) keyOrValue).length + "]"; + if (keyOrValue instanceof byte[] keyOrValueBytes) { + return "byte[" + keyOrValueBytes.length + "]"; } else { return toDisplayString(ObjectUtils.nullSafeToString(keyOrValue), 50); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index d63a5cb01..8db6a9932 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2023 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. @@ -82,6 +82,7 @@ import org.springframework.messaging.converter.MessageConverter; * @author Aldo Sinanaj * @author Chris Bono * @author Yi Liu + * @author Byungjun You */ @Configuration(proxyBeanMethods = false) @ConditionalOnMissingBean(Binder.class) @@ -200,16 +201,14 @@ public class KafkaBinderConfiguration { @Override public void configure(ProducerFactory pf) { if (pf instanceof DefaultKafkaProducerFactory) { - ((DefaultKafkaProducerFactory) pf) - .addListener(new MicrometerProducerListener<>(meterRegistry)); + pf.addListener(new MicrometerProducerListener<>(meterRegistry)); } } @Override public void configure(ConsumerFactory cf) { if (cf instanceof DefaultKafkaConsumerFactory) { - ((DefaultKafkaConsumerFactory) cf) - .addListener(new MicrometerConsumerListener<>(meterRegistry)); + cf.addListener(new MicrometerConsumerListener<>(meterRegistry)); } } @@ -255,16 +254,14 @@ public class KafkaBinderConfiguration { @Override public void configure(ProducerFactory pf) { if (pf instanceof DefaultKafkaProducerFactory) { - ((DefaultKafkaProducerFactory) pf) - .addListener(new MicrometerProducerListener<>(this.meterRegistry)); + pf.addListener(new MicrometerProducerListener<>(this.meterRegistry)); } } @Override public void configure(ConsumerFactory cf) { if (cf instanceof DefaultKafkaConsumerFactory) { - ((DefaultKafkaConsumerFactory) cf) - .addListener(new MicrometerConsumerListener<>(this.meterRegistry)); + cf.addListener(new MicrometerConsumerListener<>(this.meterRegistry)); } } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index 984d853d2..359a75d6a 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2022 the original author or authors. + * Copyright 2016-2023 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. @@ -78,6 +78,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Oleg Zhurakousky * @author Michael Michailidis + * @author Byungjun You */ // @checkstyle:off public class RabbitExchangeQueueProvisioner @@ -380,14 +381,14 @@ public class RabbitExchangeQueueProvisioner bindingKey += "-" + index; Map arguments = new HashMap<>(); arguments.putAll(extendedProperties.getQueueBindingArguments()); - if (exchange instanceof TopicExchange) { - Binding binding = BindingBuilder.bind(queue).to((TopicExchange) exchange) + if (exchange instanceof TopicExchange topicExchange) { + Binding binding = BindingBuilder.bind(queue).to(topicExchange) .with(bindingKey); declareBinding(queue.getName(), binding); return binding; } - else if (exchange instanceof DirectExchange) { - Binding binding = BindingBuilder.bind(queue).to((DirectExchange) exchange) + else if (exchange instanceof DirectExchange directExchange) { + Binding binding = BindingBuilder.bind(queue).to(directExchange) .with(bindingKey); declareBinding(queue.getName(), binding); return binding; @@ -497,8 +498,8 @@ public class RabbitExchangeQueueProvisioner : properties.getDeadLetterRoutingKey(), arguments); declareBinding(dlqName, dlqBinding); - if (properties instanceof RabbitConsumerProperties - && ((RabbitConsumerProperties) properties).isRepublishToDlq()) { + if (properties instanceof RabbitConsumerProperties rabbitConsumerProperties + && rabbitConsumerProperties.isRepublishToDlq()) { /* * Also bind with the base queue name when republishToDlq is used, which * does not know about partitioning @@ -760,9 +761,9 @@ public class RabbitExchangeQueueProvisioner true)).forEach(name -> { String group = null; String bindingName = null; - if (destination instanceof RabbitConsumerDestination) { - group = ((RabbitConsumerDestination) destination).getGroup(); - bindingName = ((RabbitConsumerDestination) destination).getBindingName(); + if (destination instanceof RabbitConsumerDestination rabbitConsumerDestination) { + group = rabbitConsumerDestination.getGroup(); + bindingName = rabbitConsumerDestination.getBindingName(); } RabbitConsumerProperties properties = consumerProperties.getExtension(); String toRemove = properties.isQueueNameGroupOnly() ? bindingName + "." + group : name.trim(); @@ -780,8 +781,8 @@ public class RabbitExchangeQueueProvisioner ExtendedProducerProperties properties) { synchronized (this.autoDeclareContext) { - if (dest instanceof RabbitProducerDestination) { - String qual = ((RabbitProducerDestination) dest).getBeanNameQualifier(); + if (dest instanceof RabbitProducerDestination rabbitProducerDestination) { + String qual = rabbitProducerDestination.getBeanNameQualifier(); removeSingleton(dest.getName() + "." + qual + ".exchange"); String[] requiredGroups = properties.getRequiredGroups(); if (!ObjectUtils.isEmpty(requiredGroups)) { @@ -854,8 +855,8 @@ public class RabbitExchangeQueueProvisioner if (this.autoDeclareContext.containsBean(name)) { ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext .getBeanFactory(); - if (beanFactory instanceof DefaultListableBeanFactory) { - ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); + if (beanFactory instanceof DefaultListableBeanFactory defaultListableBeanFactory) { + defaultListableBeanFactory.destroySingleton(name); } } } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 7d48d9adf..57d8955c4 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2022 the original author or authors. + * Copyright 2013-2023 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. @@ -134,6 +134,7 @@ import org.springframework.util.StringUtils; * @author Soby Chacko * @author Oleg Zhurakousky * @author Christian Tzolov + * @author Byungjun You */ // @checkstyle:off public class RabbitMessageChannelBinder extends @@ -271,9 +272,9 @@ public class RabbitMessageChannelBinder extends @Override public void destroy() throws Exception { - if (this.connectionFactory instanceof DisposableBean) { + if (this.connectionFactory instanceof DisposableBean disposableConnectionFactory) { if (this.destroyConnectionFactory) { - ((DisposableBean) this.connectionFactory).destroy(); + disposableConnectionFactory.destroy(); } } } @@ -968,11 +969,9 @@ public class RabbitMessageChannelBinder extends } } else { - if (message.getPayload() instanceof MessagingException) { + if (message.getPayload() instanceof MessagingException messagingException) { AcknowledgmentCallback ack = StaticMessageHeaderAccessor - .getAcknowledgmentCallback( - ((MessagingException) message.getPayload()) - .getFailedMessage()); + .getAcknowledgmentCallback(messagingException.getFailedMessage()); if (ack != null) { if (properties.getExtension().isRequeueRejected()) { ack.acknowledge(Status.REQUEUE); @@ -1088,8 +1087,8 @@ public class RabbitMessageChannelBinder extends @Override protected Message createMessage(Object object, MessageProperties messageProperties) { - if (object instanceof byte[]) { - return new Message((byte[]) object, messageProperties); + if (object instanceof byte[] bytes) { + return new Message(bytes, messageProperties); } else { // just for safety (backwards compatibility) diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ObservationAutoConfiguration.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ObservationAutoConfiguration.java index e081e3522..60729f028 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ObservationAutoConfiguration.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/ObservationAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2022 the original author or authors. + * Copyright 2022-2023 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. @@ -29,6 +29,10 @@ import org.springframework.core.annotation.Order; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; import org.springframework.messaging.MessageHandler; +/** + * @author Oleg Zhurakousky + * @author Byungjun You + */ @Configuration(proxyBeanMethods = false) @ConditionalOnBean(org.springframework.boot.actuate.autoconfigure.observation.ObservationAutoConfiguration.class) public class ObservationAutoConfiguration { @@ -50,9 +54,9 @@ public class ObservationAutoConfiguration { ProducerMessageHandlerCustomizer observedProducerMessageHandlerCustomizer( ApplicationContext applicationContext) { return (handler, destinationName) -> { - if (handler instanceof AmqpOutboundEndpoint) { - ((AmqpOutboundEndpoint) handler).getRabbitTemplate().setObservationEnabled(true); - ((AmqpOutboundEndpoint) handler).getRabbitTemplate().setApplicationContext(applicationContext); + if (handler instanceof AmqpOutboundEndpoint amqpOutboundEndpoint) { + amqpOutboundEndpoint.getRabbitTemplate().setObservationEnabled(true); + amqpOutboundEndpoint.getRabbitTemplate().setApplicationContext(applicationContext); } }; } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index c43eeb2fd..1ffb084f9 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2023 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. @@ -58,6 +58,7 @@ import org.springframework.util.CollectionUtils; * @author Oleg Zhurakousky * @author Gary Russell * @author Ben Blinebury + * @author Byungjun You */ @Configuration(proxyBeanMethods = false) @Import({ PropertyPlaceholderAutoConfiguration.class }) @@ -87,9 +88,10 @@ public class RabbitMessageChannelBinderConfiguration { @Nullable ConnectionNameStrategy connectionNameStrategy) { String connectionNamePrefix = this.rabbitBinderConfigurationProperties.getConnectionNamePrefix(); - if (this.rabbitConnectionFactory instanceof AbstractConnectionFactory && connectionNamePrefix != null && connectionNameStrategy == null) { + if (this.rabbitConnectionFactory instanceof AbstractConnectionFactory connectionFactory + && connectionNamePrefix != null && connectionNameStrategy == null) { final AtomicInteger nameIncrementer = new AtomicInteger(); - ((AbstractConnectionFactory) this.rabbitConnectionFactory).setConnectionNameStrategy(f -> connectionNamePrefix + connectionFactory.setConnectionNameStrategy(f -> connectionNamePrefix + "#" + nameIncrementer.getAndIncrement()); } diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index 451c91a48..41063edb0 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2022 the original author or authors. + * Copyright 2015-2023 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. @@ -94,6 +94,7 @@ import static org.mockito.Mockito.verify; * @author Artem Bilan * @author Soby Chacko * @author Chris Bono + * @author Byungjun You */ public class RabbitBinderModuleTests { @@ -436,8 +437,8 @@ public class RabbitBinderModuleTests { @Bean public ProducerMessageHandlerCustomizer messageHandlerCustomizer() { return (handler, destinationName) -> { - if (handler instanceof AbstractMessageHandler) { - ((AbstractMessageHandler) handler).setBeanName("setByCustomizer:" + destinationName); + if (handler instanceof AbstractMessageHandler messageHandler) { + messageHandler.setBeanName("setByCustomizer:" + destinationName); } }; } diff --git a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java index e3bc3417f..cf330201c 100644 --- a/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java +++ b/core/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2022 the original author or authors. + * Copyright 2017-2023 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. @@ -57,6 +57,7 @@ import static org.assertj.core.api.Assertions.assertThat; * * @author Oleg Zhurakousky * @author Gary Russell + * @author Byungjun You * */ public class ContentTypeTckTests { @@ -586,8 +587,8 @@ public class ContentTypeTckTests { @Bean public Function echo(JsonMapper mapper) { - return value -> value instanceof byte[] - ? mapper.fromJson((byte[]) value, Person.class) + return value -> value instanceof byte[] valueAsBytes + ? mapper.fromJson(valueAsBytes, Person.class) : mapper.fromJson((String) value, Person.class); } } @@ -598,8 +599,8 @@ public class ContentTypeTckTests { @Bean public Function, Person> echo(JsonMapper mapper) { - return message -> message.getPayload() instanceof byte[] - ? mapper.fromJson((byte[]) message.getPayload(), Person.class) + return message -> message.getPayload() instanceof byte[] payloadAsBytes + ? mapper.fromJson(payloadAsBytes, Person.class) : mapper.fromJson((String) message.getPayload(), Person.class); } } diff --git a/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java b/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java index d34512e3b..d6938fcf0 100644 --- a/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java +++ b/core/spring-cloud-stream-test-binder/src/main/java/org/springframework/cloud/stream/binder/test/FunctionBindingTestUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2023 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. @@ -39,6 +39,7 @@ import org.springframework.messaging.MessageChannel; * It is intended for internal framework testing and is NOT for public use. Breaking changes are likely!!! * * @author Oleg Zhurakousky + * @author Byungjun You * @since 3.0.2 * */ @@ -52,8 +53,8 @@ public final class FunctionBindingTestUtils { public static void bind(ConfigurableApplicationContext applicationContext, Object function) { try { Object targetFunction = function; - if (function instanceof FunctionRegistration) { - targetFunction = ((FunctionRegistration) function).getTarget(); + if (function instanceof FunctionRegistration functionRegistration) { + targetFunction = functionRegistration.getTarget(); } String functionName = targetFunction instanceof Function ? "function" : (targetFunction instanceof Consumer ? "consumer" : "supplier"); diff --git a/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java b/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java index 41d3e1325..cb5eff10e 100644 --- a/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java +++ b/core/spring-cloud-stream-test-support/src/main/java/org/springframework/cloud/stream/binder/AbstractTestBinder.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2023 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. @@ -34,6 +34,7 @@ import org.springframework.util.CollectionUtils; * @author Gary Russell * @author Mark Fisher * @author Oleg Zhurakousky + * @author Byungjun You */ // @checkstyle:off public abstract class AbstractTestBinder, CP extends ConsumerProperties, PP extends ProducerProperties> @@ -86,12 +87,11 @@ public abstract class AbstractTestBinder> @@ -268,16 +268,16 @@ public abstract class AbstractMessageChannelBinder reactiveStreamsConsumerRef = new AtomicReference<>(); - if (outputChannel instanceof SubscribableChannel) { - ((SubscribableChannel) outputChannel) + if (outputChannel instanceof SubscribableChannel subscribableOutputChannel) { + subscribableOutputChannel .subscribe(new SendingHandler(producerMessageHandler, HeaderMode.embeddedHeaders .equals(producerProperties.getHeaderMode()), @@ -310,8 +310,8 @@ public abstract class AbstractMessageChannelBinder binding = new DefaultBinding(destination, - outputChannel, producerMessageHandler instanceof Lifecycle - ? (Lifecycle) producerMessageHandler : null) { + outputChannel, producerMessageHandler instanceof Lifecycle producerMessageHandlerWithLifecycle + ? producerMessageHandlerWithLifecycle : null) { @Override public Map getExtendedInfo() { @@ -331,8 +331,8 @@ public abstract class AbstractMessageChannelBinder binding = new DefaultBinding(name, - group, inputChannel, consumerEndpoint instanceof Lifecycle - ? (Lifecycle) consumerEndpoint : null) { + group, inputChannel, consumerEndpoint instanceof Lifecycle consumerEndpointWithLifecycle + ? consumerEndpointWithLifecycle : null) { @Override public Map getExtendedInfo() { @@ -498,8 +498,8 @@ public abstract class AbstractMessageChannelBinder messageSource = resources.getSource(); - if (messageSource instanceof BeanFactoryAware) { - ((BeanFactoryAware) messageSource).setBeanFactory(getApplicationContext().getBeanFactory()); + if (messageSource instanceof BeanFactoryAware beanFactoryAwareMessageSource) { + beanFactoryAwareMessageSource.setBeanFactory(getApplicationContext().getBeanFactory()); } bindingTarget.setSource(messageSource); if (resources.getErrorInfrastructure() != null) { @@ -565,12 +565,12 @@ public abstract class AbstractMessageChannelBinder> binding = new DefaultBinding>( - name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle - ? (Lifecycle) resources.getSource() : null) { + name, group, inboundBindTarget, resources.getSource() instanceof Lifecycle sourceWithLifecycle + ? sourceWithLifecycle : null) { @Override public Map getExtendedInfo() { @@ -818,9 +818,8 @@ public abstract class AbstractMessageChannelBinder preSend(Message message, MessageChannel channel) { - if (message.getPayload() instanceof byte[] + if (message.getPayload() instanceof byte[] messagePayload && !message.getHeaders() .containsKey(BinderHeaders.NATIVE_HEADERS_PRESENT) && EmbeddedHeaderUtils - .mayHaveEmbeddedHeaders((byte[]) message.getPayload())) { + .mayHaveEmbeddedHeaders(messagePayload)) { MessageValues messageValues; try { @@ -1140,22 +1139,22 @@ public abstract class AbstractMessageChannelBinder messageConverters = binderProducingContext.getBeansOfType(MessageConverter.class); if (!CollectionUtils.isEmpty(messageConverters) && !ObjectUtils.isEmpty(context.getBeansOfType(FunctionCatalog.class))) { FunctionCatalog functionCatalog = this.context.getBean(FunctionCatalog.class); - if (functionCatalog instanceof SimpleFunctionRegistry) { - ((SimpleFunctionRegistry) functionCatalog).addMessageConverters(messageConverters.values()); + if (functionCatalog instanceof SimpleFunctionRegistry simpleFunctionRegistry) { + simpleFunctionRegistry.addMessageConverters(messageConverters.values()); } } @@ -323,8 +324,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl * accessible within binder's context (see * https://github.com/spring-cloud/spring-cloud-stream/issues/1384) */ - if (this.context != null && binder instanceof ApplicationContextAware) { - ((ApplicationContextAware) binder).setApplicationContext(this.context); + if (this.context != null && binder instanceof ApplicationContextAware applicationContextAwareBinder) { + applicationContextAwareBinder.setApplicationContext(this.context); } if (!CollectionUtils.isEmpty(this.listeners)) { for (Listener binderFactoryListener : this.listeners) { @@ -555,12 +556,12 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl * foo={bar=baz}). * @param propertyName property name to flatten * @param value value that contains the property name - * @param flattenedProperties map to which we'll add the falttened property + * @param flattenedProperties map to which we'll add the flattened property */ @SuppressWarnings("unchecked") private void flatten(String propertyName, Object value, Map flattenedProperties) { - if (value instanceof Map) { - ((Map) value).forEach((k, v) -> flatten( + if (value instanceof Map valueAsMap) { + valueAsMap.forEach((k, v) -> flatten( (propertyName != null ? propertyName + "." : "") + k, v, flattenedProperties)); } else { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java index e3d7c0387..c9e382ddc 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2021 the original author or authors. + * Copyright 2013-2023 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. @@ -40,6 +40,7 @@ import org.springframework.util.StringUtils; * @author Oleg Zhurakousky * @author Myeonghyeon Lee * @author Soby Chacko + * @author Byungjun You * @see org.springframework.cloud.stream.annotation.EnableBinding */ @@ -93,8 +94,8 @@ public class DefaultBinding implements Binding { @Override public String getBindingName() { - String resolvedName = (this.target instanceof IntegrationObjectSupport) - ? ((IntegrationObjectSupport) this.target).getComponentName() : getName(); + String resolvedName = (this.target instanceof IntegrationObjectSupport integrationObjectSupportTarget) + ? integrationObjectSupportTarget.getComponentName() : getName(); return resolvedName == null ? getName() : resolvedName; } @@ -156,8 +157,8 @@ public class DefaultBinding implements Binding { @Override public synchronized void pause() { - if (this.lifecycle instanceof Pausable) { - ((Pausable) this.lifecycle).pause(); + if (this.lifecycle instanceof Pausable pausableLifecycle) { + pausableLifecycle.pause(); this.paused = true; } else { @@ -169,8 +170,8 @@ public class DefaultBinding implements Binding { @Override public synchronized void resume() { - if (this.lifecycle instanceof Pausable) { - ((Pausable) this.lifecycle).resume(); + if (this.lifecycle instanceof Pausable pausableLifecycle) { + pausableLifecycle.resume(); this.paused = false; } else { @@ -193,8 +194,8 @@ public class DefaultBinding implements Binding { @Override public String toString() { return " Binding [name=" + this.name + ", target=" + this.target + ", lifecycle=" - + ((this.lifecycle instanceof NamedComponent) - ? ((NamedComponent) this.lifecycle).getComponentName() + + ((this.lifecycle instanceof NamedComponent namedComponentWithLifeCycle) + ? namedComponentWithLifeCycle.getComponentName() : ObjectUtils.nullSafeToString(this.lifecycle)) + "]"; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index cd3a35c80..4788d99b4 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2020 the original author or authors. + * Copyright 2018-2023 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. @@ -62,6 +62,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Oleg Zhurakousky * @author David Turanski + * @author Byungjun You * @since 2.0 * */ @@ -115,8 +116,7 @@ public class DefaultPollableMessageSource @Override public Object invoke(MethodInvocation invocation) throws Throwable { Object result = invocation.proceed(); - if (result instanceof Message) { - Message received = (Message) result; + if (result instanceof Message received) { for (ChannelInterceptor interceptor : this.interceptors) { received = interceptor.preSend(received, dummyChannel); if (received == null) { @@ -181,16 +181,16 @@ public class DefaultPollableMessageSource @Override public synchronized void start() { - if (!this.running && this.source instanceof Lifecycle) { - ((Lifecycle) this.source).start(); + if (!this.running && this.source instanceof Lifecycle sourceWithLifecycle) { + sourceWithLifecycle.start(); } this.running = true; } @Override public synchronized void stop() { - if (this.running && this.source instanceof Lifecycle) { - ((Lifecycle) this.source).stop(); + if (this.running && this.source instanceof Lifecycle sourceWithLifeCycle) { + sourceWithLifeCycle.stop(); } this.running = false; } @@ -244,8 +244,8 @@ public class DefaultPollableMessageSource } catch (Exception e) { AckUtils.autoNack(ackCallback); - if (e instanceof MessageHandlingException && ((MessageHandlingException) e) - .getFailedMessage().equals(message)) { + if (e instanceof MessageHandlingException messageHandlingException && + messageHandlingException.getFailedMessage().equals(message)) { throw (MessageHandlingException) e; } throw new MessageHandlingException(message, e); @@ -309,7 +309,7 @@ public class DefaultPollableMessageSource Message message = this.source.receive(); if (message != null && type != null && this.messageConverter != null) { Class targetType = type == null ? Object.class - : type.getType() instanceof Class ? (Class) type.getType() + : type.getType() instanceof Class clazz ? clazz : Object.class; Object payload = this.messageConverter.fromMessage(message, targetType, type); if (payload == null) { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java index b742f3fa8..7d2531c22 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/FinalRethrowingErrorMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2023 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. @@ -27,6 +27,7 @@ import org.springframework.messaging.MessagingException; * handled from {@code getErrorMessageHandler()}. * * @author Gary Russell + * @author Byungjun You * @since 1.3 * */ @@ -35,8 +36,8 @@ class FinalRethrowingErrorMessageHandler @Override public void handleMessage(Message message) throws MessagingException { - if (message.getPayload() instanceof MessagingException) { - throw (MessagingException) message.getPayload(); + if (message.getPayload() instanceof MessagingException messagingException) { + throw messagingException; } else { throw new MessagingException((Message) null, diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index ffa70a6ac..8abc02922 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -62,6 +62,7 @@ import org.springframework.validation.DataBinder; * @author Michael Michailidis * @author Chris Bono * @author Artem Bilan + * @author Byungjun You */ public class BindingService { @@ -100,16 +101,16 @@ public class BindingService { public Collection> bindConsumer(T input, String inputName) { Collection> bindings = new ArrayList<>(); Class inputClass = input.getClass(); - if (input instanceof Advised) { - inputClass = Stream.of(((Advised) input).getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() + if (input instanceof Advised advisedInput) { + inputClass = Stream.of(advisedInput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() .orElse(inputClass); } Binder binder = (Binder) getBinder( inputName, inputClass); ConsumerProperties consumerProperties = this.bindingServiceProperties .getConsumerProperties(inputName); - if (binder instanceof ExtendedPropertiesBinder) { - Object extension = ((ExtendedPropertiesBinder) binder) + if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { + Object extension = extendedPropertiesBinder .getExtendedConsumerProperties(inputName); ExtendedConsumerProperties extendedConsumerProperties = new ExtendedConsumerProperties( extension); @@ -146,11 +147,11 @@ public class BindingService { continue; } - Object extensiion = consumerProperties instanceof ExtendedConsumerProperties - ? ((ExtendedConsumerProperties) consumerProperties).getExtension() + Object extension = consumerProperties instanceof ExtendedConsumerProperties extendedProperties + ? extendedProperties.getExtension() : null; - ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>(extensiion); + ConsumerProperties consumerPropertiesTemp = new ExtendedConsumerProperties<>(extension); BeanUtils.copyProperties(consumerProperties, consumerPropertiesTemp); consumerPropertiesTemp.setInstanceIndex(index); @@ -270,8 +271,8 @@ public class BindingService { public Binding bindProducer(T output, String outputName, boolean cache, @Nullable Binder binder) { String bindingTarget = this.bindingServiceProperties.getBindingDestination(outputName); Class outputClass = output.getClass(); - if (output instanceof Advised) { - outputClass = Stream.of(((Advised) output).getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() + if (output instanceof Advised advisedOutput) { + outputClass = Stream.of(advisedOutput.getProxiedInterfaces()).filter(c -> !c.getName().contains("org.springframework")).findFirst() .orElse(outputClass); } if (binder == null) { @@ -280,9 +281,8 @@ public class BindingService { ProducerProperties producerProperties = this.bindingServiceProperties .getProducerProperties(outputName); - if (binder instanceof ExtendedPropertiesBinder) { - Object extension = ((ExtendedPropertiesBinder) binder) - .getExtendedProducerProperties(outputName); + if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { + Object extension = extendedPropertiesBinder.getExtendedProducerProperties(outputName); ExtendedProducerProperties extendedProducerProperties = new ExtendedProducerProperties<>( extension); BeanUtils.copyProperties(producerProperties, extendedProducerProperties); @@ -310,9 +310,8 @@ public class BindingService { @SuppressWarnings("rawtypes") public Object getExtendedProducerProperties(Object output, String outputName) { Binder binder = getBinder(outputName, output.getClass()); - if (binder instanceof ExtendedPropertiesBinder) { - return ((ExtendedPropertiesBinder) binder) - .getExtendedProducerProperties(outputName); + if (binder instanceof ExtendedPropertiesBinder extendedPropertiesBinder) { + return extendedPropertiesBinder.getExtendedProducerProperties(outputName); } return null; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java index 93830eb72..5452a30af 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/SubscribableChannelBindingTargetFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2023 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. @@ -30,6 +30,7 @@ import org.springframework.messaging.SubscribableChannel; * @author David Syer * @author Ilayaperumal Gopinathan * @author Oleg Zhurakousky + * @author Byungjun You */ public class SubscribableChannelBindingTargetFactory extends AbstractBindingTargetFactory { @@ -68,9 +69,9 @@ public class SubscribableChannelBindingTargetFactory } subscribableChannel = channel; } - if (subscribableChannel instanceof DirectWithAttributesChannel) { - ((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "input"); - this.messageChannelConfigurer.configureInputChannel(subscribableChannel, name); + if (subscribableChannel instanceof DirectWithAttributesChannel directWithAttributesChannel) { + directWithAttributesChannel.setAttribute("type", "input"); + this.messageChannelConfigurer.configureInputChannel(directWithAttributesChannel, name); } return subscribableChannel; @@ -95,9 +96,9 @@ public class SubscribableChannelBindingTargetFactory } subscribableChannel = channel; } - if (subscribableChannel instanceof DirectWithAttributesChannel) { - ((DirectWithAttributesChannel) subscribableChannel).setAttribute("type", "output"); - this.messageChannelConfigurer.configureOutputChannel(subscribableChannel, name); + if (subscribableChannel instanceof DirectWithAttributesChannel directWithAttributesChannel) { + directWithAttributesChannel.setAttribute("type", "output"); + this.messageChannelConfigurer.configureOutputChannel(directWithAttributesChannel, name); } return subscribableChannel; diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java index 37520cd88..4d24384ff 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ObjectStringMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2023 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. @@ -35,6 +35,7 @@ import org.springframework.util.MimeType; * * @author Marius Bogoevici * @author Oleg Zhurakousky + * @author Byungjun You * @since 1.2 */ public class ObjectStringMessageConverter extends AbstractMessageConverter { @@ -74,30 +75,28 @@ public class ObjectStringMessageConverter extends AbstractMessageConverter { // Assert.isTrue(String.class.isAssignableFrom(targetClass) || targetClass == Object.class, "This converter can only convert byte[] to String"); if (message.getPayload() != null) { - if (message.getPayload() instanceof byte[]) { + if (message.getPayload() instanceof byte[] payloadAsBytes) { if (byte[].class.isAssignableFrom(targetClass)) { return message.getPayload(); } else { - return new String((byte[]) message.getPayload(), + return new String(payloadAsBytes, StandardCharsets.UTF_8); } } - else if (message.getPayload() instanceof Collection) { - Collection collection = ((Collection) message.getPayload()).stream() + else if (message.getPayload() instanceof Collection payloadAsCollection) { + return payloadAsCollection.stream() .map(value -> { if (byte[].class.isAssignableFrom(targetClass)) { return value; } - else if (value instanceof byte[]) { - return new String((byte[]) value, StandardCharsets.UTF_8); + else if (value instanceof byte[] valueAsBytes) { + return new String(valueAsBytes, StandardCharsets.UTF_8); } else { return value; // String } }).collect(Collectors.toList()); - - return collection; } else { if (byte[].class.isAssignableFrom(targetClass)) { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 91d996929..70960730d 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-2023 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. @@ -119,6 +119,7 @@ import org.springframework.util.StringUtils; * @author Ilayaperumal Gopinathan * @author Soby Chacko * @author Chris Bono + * @author Byungjun You * @since 2.1 */ @AutoConfiguration @@ -297,8 +298,9 @@ public class FunctionConfiguration { boolean splittable = pollable != null && (boolean) AnnotationUtils.getAnnotationAttributes(pollable).get("splittable"); - FunctionInvocationWrapper function = (supplier instanceof PartitionAwareFunctionWrapper) - ? (FunctionInvocationWrapper) ((PartitionAwareFunctionWrapper) supplier).function : (FunctionInvocationWrapper) supplier; + FunctionInvocationWrapper function = + (supplier instanceof PartitionAwareFunctionWrapper partitionAwareFunctionWrapper) + ? (FunctionInvocationWrapper) partitionAwareFunctionWrapper.function : (FunctionInvocationWrapper) supplier; boolean reactive = FunctionTypeUtils.isPublisher(function.getOutputType()); if (pollable == null && reactive) { @@ -361,17 +363,17 @@ public class FunctionConfiguration { String supplierFunctionName = StringUtils .delimitedListToStringArray(proxyFactory.getFunctionDefinition().replaceAll(",", "|").trim(), "|")[0]; BeanDefinition bd = context.getBeanDefinition(supplierFunctionName); - if (!(bd instanceof RootBeanDefinition)) { + if (!(bd instanceof RootBeanDefinition rootBeanDefinition)) { return null; } - Method factoryMethod = ((RootBeanDefinition) bd).getResolvedFactoryMethod(); + Method factoryMethod = rootBeanDefinition.getResolvedFactoryMethod(); if (factoryMethod == null) { Object source = bd.getSource(); - if (source instanceof MethodMetadata) { - Class factory = ClassUtils.resolveClassName(((MethodMetadata) source).getDeclaringClassName(), null); + if (source instanceof MethodMetadata methodMetadata) { + Class factory = ClassUtils.resolveClassName(methodMetadata.getDeclaringClassName(), null); Class[] params = FunctionContextUtils.getParamTypesFromBeanDefinitionFactory(factory, (RootBeanDefinition) bd); - factoryMethod = ReflectionUtils.findMethod(factory, ((MethodMetadata) source).getMethodName(), params); + factoryMethod = ReflectionUtils.findMethod(factory, methodMetadata.getMethodName(), params); } } Assert.notNull(factoryMethod, "Failed to introspect factory method since it was not discovered for function '" @@ -384,8 +386,8 @@ public class FunctionConfiguration { @SuppressWarnings("unchecked") private Message wrapToMessageIfNecessary(T value) { - return value instanceof Message - ? (Message) value + return value instanceof Message message + ? message : MessageBuilder.withPayload(value).build(); } @@ -436,8 +438,9 @@ public class FunctionConfiguration { Map beansOfType = applicationContext.getBeansOfType(BindableProxyFactory.class); this.bindableProxyFactories = beansOfType.values().toArray(new BindableProxyFactory[0]); for (BindableProxyFactory bindableProxyFactory : this.bindableProxyFactories) { - String functionDefinition = bindableProxyFactory instanceof BindableFunctionProxyFactory && ((BindableFunctionProxyFactory) bindableProxyFactory).isFunctionExist() - ? ((BindableFunctionProxyFactory) bindableProxyFactory).getFunctionDefinition() + String functionDefinition = bindableProxyFactory instanceof BindableFunctionProxyFactory functionFactory + && functionFactory.isFunctionExist() + ? functionFactory.getFunctionDefinition() : null; /*this.functionProperties.getDefinition();*/ boolean shouldNotProcess = false; @@ -558,17 +561,14 @@ public class FunctionConfiguration { if (!CollectionUtils.isEmpty(outputBindingNames)) { MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class); flux = flux.doOnNext(message -> { - if (message instanceof Message && ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { - String destinationName = (String) ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination"); + if (message instanceof Message m && m.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { + String destinationName = (String) m.getHeaders().get("spring.cloud.stream.sendto.destination"); ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer(); MessageChannel dynamicChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } - if (!(message instanceof Message)) { - message = MessageBuilder.withPayload(message).build(); - } - dynamicChannel.send((Message) message); + dynamicChannel.send(m); } else { if (!(message instanceof Message)) { @@ -623,8 +623,8 @@ public class FunctionConfiguration { logger.debug("Function execution resulted in null. No message will be sent"); return; } - if (result instanceof Iterable) { - for (Object resultElement : (Iterable) result) { + if (result instanceof Iterable iterableResult) { + for (Object resultElement : iterableResult) { this.doSendMessage(resultElement, message); } } @@ -639,13 +639,13 @@ public class FunctionConfiguration { } private void doSendMessage(Object result, Message requestMessage) { - if (result instanceof Message && ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { - String destinationName = (String) ((Message) result).getHeaders().get("spring.cloud.stream.sendto.destination"); + if (result instanceof Message messageResult && messageResult.getHeaders().get("spring.cloud.stream.sendto.destination") != null) { + String destinationName = (String) messageResult.getHeaders().get("spring.cloud.stream.sendto.destination"); MessageChannel outputChannel = streamBridge.resolveDestination(destinationName, producerProperties, null); if (logger.isInfoEnabled()) { logger.info("Output message is sent to '" + destinationName + "' destination"); } - outputChannel.send(((Message) result)); + outputChannel.send(messageResult); } else if (StringUtils.hasText(outputChannelName)) { if (!(result instanceof Message)) { diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java index e6cfbe5f6..0a4bdbd8d 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/PartitionAwareFunctionWrapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 the original author or authors. + * Copyright 2020-2023 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. @@ -37,6 +37,10 @@ import org.springframework.util.ObjectUtils; * This class is effectively a wrapper which is aware of the stream related partition information * for outgoing messages. It has only one responsibility and that is to modify the result message * with 'scst_partition' header if necessary. + * + * @author Oleg Zhurakousky + * @author Soby Chacko + * @author Byungjun You */ class PartitionAwareFunctionWrapper implements Function, Supplier { @@ -90,16 +94,16 @@ class PartitionAwareFunctionWrapper implements Function, Supplie @Override public Object get() { - if (this.function instanceof FunctionInvocationWrapper) { + if (this.function instanceof FunctionInvocationWrapper functionInvocationWrapper) { this.setEnhancerIfNecessary(); - return ((FunctionInvocationWrapper) this.function).get(); + return functionInvocationWrapper.get(); } throw new IllegalStateException("Call to get() is not allowed since this function is not a Supplier."); } private void setEnhancerIfNecessary() { - if (this.function instanceof FunctionInvocationWrapper) { - ((FunctionInvocationWrapper) this.function).setEnhancer(this.outputMessageEnricher); + if (this.function instanceof FunctionInvocationWrapper functionInvocationWrapper) { + functionInvocationWrapper.setEnhancer(this.outputMessageEnricher); } } } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java index e3dbe1c16..f445d7105 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java @@ -72,6 +72,7 @@ import org.springframework.util.StringUtils; * * @author Oleg Zhurakousky * @author Soby Chacko + * @author Byungjun You * @since 3.0.3 * */ @@ -164,8 +165,8 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi String targetType = this.resolveBinderTargetType(bindingName, binderName, MessageChannel.class, this.applicationContext.getBean(BinderFactory.class)); - Message messageToSend = data instanceof Message - ? MessageBuilder.fromMessage((Message) data).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build() + Message messageToSend = data instanceof Message messageData + ? MessageBuilder.fromMessage(messageData).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build() : new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, targetType)); Message resultMessage; diff --git a/docs/src/main/asciidoc/kafka/kafka_overview.adoc b/docs/src/main/asciidoc/kafka/kafka_overview.adoc index dca41b052..e82a5b3c1 100644 --- a/docs/src/main/asciidoc/kafka/kafka_overview.adoc +++ b/docs/src/main/asciidoc/kafka/kafka_overview.adoc @@ -944,8 +944,8 @@ to apply, you need to use a 'BinderCustomizer' to set the container customizer ( @Bean public BinderCustomizer binderCustomizer(ListenerContainerWithDlqAndRetryCustomizer containerCustomizer) { return (binder, binderName) -> { - if (binder instanceof KafkaMessageChannelBinder) { - ((KafkaMessageChannelBinder) binder).setContainerCustomizer(containerCustomizer); + if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) { + kafkaMessageChannelBinder.setContainerCustomizer(containerCustomizer); } else if (binder instanceof KStreamBinder) { ... diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 019099ea8..b38df534c 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -1879,8 +1879,8 @@ Here is an example of providing a `BinderCustomizer` bean. @Bean public BinderCustomizer binderCustomizer() { return (binder, binderName) -> { - if (binder instanceof KafkaMessageChannelBinder) { - ((KafkaMessageChannelBinder) binder).setRebalanceListener(...); + if (binder instanceof KafkaMessageChannelBinder kafkaMessageChannelBinder) { + kafkaMessageChannelBinder.setRebalanceListener(...); } else if (binder instanceof KStreamBinder) { ... diff --git a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AbstractAvroMessageConverter.java b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AbstractAvroMessageConverter.java index 95ca6b846..95df97f50 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AbstractAvroMessageConverter.java +++ b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/AbstractAvroMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2018 the original author or authors. + * Copyright 2016-2023 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. @@ -41,6 +41,7 @@ import org.springframework.util.MimeType; * @author Vinicius Carvalho * @author Sercan Karaoglu * @author Ish Mahajan + * @author Byungjun You */ public abstract class AbstractAvroMessageConverter extends AbstractMessageConverter { @@ -93,8 +94,8 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver MimeType mimeType = getContentTypeResolver().resolve(message.getHeaders()); if (mimeType == null) { - if (conversionHint instanceof MimeType) { - mimeType = (MimeType) conversionHint; + if (conversionHint instanceof MimeType hintedMimeType) { + mimeType = hintedMimeType; } else { return null; @@ -117,8 +118,8 @@ public abstract class AbstractAvroMessageConverter extends AbstractMessageConver ByteArrayOutputStream baos = new ByteArrayOutputStream(); try { MimeType hintedContentType = null; - if (conversionHint instanceof MimeType) { - hintedContentType = (MimeType) conversionHint; + if (conversionHint instanceof MimeType mimeType) { + hintedContentType = mimeType; } Schema schema = resolveSchemaForWriting(payload, headers, hintedContentType); @SuppressWarnings("unchecked") diff --git a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/OriginalContentTypeResolver.java b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/OriginalContentTypeResolver.java index 8274e12da..2f3ed02e6 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/OriginalContentTypeResolver.java +++ b/schema-registry/spring-cloud-stream-schema-registry-client/src/main/java/org/springframework/cloud/stream/schema/registry/avro/OriginalContentTypeResolver.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2023 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. @@ -25,6 +25,7 @@ import org.springframework.util.MimeType; /** * @author Vinicius Carvalho + * @author Byungjun You * * Resolves contentType looking for a originalContentType header first. If not found * returns the contentType @@ -43,13 +44,12 @@ class OriginalContentTypeResolver implements ContentTypeResolver { ? headers.get(BINDER_ORIGINAL_CONTENT_TYPE) : headers.get(MessageHeaders.CONTENT_TYPE); MimeType mimeType = null; - if (contentType instanceof MimeType) { - mimeType = (MimeType) contentType; + if (contentType instanceof MimeType mimeContentType) { + mimeType = mimeContentType; } - else if (contentType instanceof String) { + else if (contentType instanceof String valueAsString) { mimeType = this.mimeTypeCache.get(contentType); if (mimeType == null) { - String valueAsString = (String) contentType; mimeType = MimeType.valueOf(valueAsString); this.mimeTypeCache.put(valueAsString, mimeType); } diff --git a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/config/SchemaServerConfiguration.java b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/config/SchemaServerConfiguration.java index 53b0fcb84..af4b70b50 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/config/SchemaServerConfiguration.java +++ b/schema-registry/spring-cloud-stream-schema-registry-core/src/main/java/org/springframework/cloud/stream/schema/registry/config/SchemaServerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2023 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. @@ -37,6 +37,7 @@ import org.springframework.data.jpa.repository.config.EnableJpaRepositories; /** * @author Vinicius Carvalho * @author Soby Chacko + * @author Byungjun You */ @Configuration(proxyBeanMethods = false) @EnableJpaRepositories(basePackageClasses = SchemaRepository.class) @@ -47,8 +48,8 @@ public class SchemaServerConfiguration { @Bean public static BeanFactoryPostProcessor entityScanPackagesPostProcessor() { return beanFactory -> { - if (beanFactory instanceof BeanDefinitionRegistry) { - EntityScanPackages.register((BeanDefinitionRegistry) beanFactory, + if (beanFactory instanceof BeanDefinitionRegistry beanDefinitionRegistry) { + EntityScanPackages.register(beanDefinitionRegistry, Collections.singletonList(Schema.class.getPackage().getName())); } };