GH-10083: Migrate ZeroMQ module to Jspecify

Related to: #10083

- Replaced `org.springframework.lang.Nullable` with `org.jspecify.annotations.Nullable`

* Complete ZeroMQ module migration to JSpecify

- Migrate `package-info.java` files to use `@NullMarked` annotation
- Add `@SuppressWarnings("NullAway.Init")` for fields initialized in lifecycle methods

* Fix NullAway violations in ZeroMqChannel

- Extract `this.zeroMqProxy` to local variable inside `Mono.defer()` block

Signed-off-by: Jooyoung Pyoung <[email protected]>

Author: anthologia

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/ZeroMqProxy.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2023 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -26,6 +26,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZMQ;
@@ -35,7 +36,6 @@
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
-import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -82,8 +82,10 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 
 	private final AtomicInteger backendPort = new AtomicInteger();
 
+	@SuppressWarnings("NullAway.Init")
 	private String controlAddress;
 
+	@SuppressWarnings("NullAway.Init")
 	private Executor proxyExecutor;
 
 	@Nullable
@@ -97,6 +99,7 @@ public class ZeroMqProxy implements InitializingBean, SmartLifecycle, BeanNameAw
 	@Nullable
 	private String captureAddress;
 
+	@SuppressWarnings("NullAway.Init")
 	private String beanName;
 
 	private boolean autoStartup = true;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/ZeroMqChannel.java

@@ -23,6 +23,7 @@
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import org.jspecify.annotations.Nullable;
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZMQ;
@@ -36,7 +37,6 @@
 import org.springframework.integration.mapping.BytesMessageMapper;
 import org.springframework.integration.support.json.EmbeddedJsonHeadersMessageMapper;
 import org.springframework.integration.zeromq.ZeroMqProxy;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.SubscribableChannel;
@@ -66,6 +66,7 @@
  * concurrency primitives for multi-publisher(subscriber) communication within the same application.
  *
  * @author Artem Bilan
+ * @author Jooyoung Pyoung
  *
  * @since 5.4
  */
@@ -142,19 +143,19 @@ public ZeroMqChannel(ZContext context, boolean pubSub) {
 		this.subscriberData = prepareSubscriberDataFlux();
 	}
 
-	@SuppressWarnings("this-escape")
 	private Mono<Integer> prepareProxyMono() {
 		return Mono.defer(() -> {
-					if (this.zeroMqProxy != null) {
-						return Mono.fromCallable(() -> this.zeroMqProxy.getBackendPort())
+					ZeroMqProxy zeroMqProxyToUse = this.zeroMqProxy;
+					if (zeroMqProxyToUse != null) {
+						return Mono.fromCallable(zeroMqProxyToUse::getBackendPort)
 								.filter((proxyPort) -> proxyPort > 0)
 								.repeatWhenEmpty(100, (repeat) -> repeat.delayElements(Duration.ofMillis(100))) // NOSONAR
 								.doOnNext((proxyPort) ->
-										setConnectUrl("tcp://localhost:" + this.zeroMqProxy.getFrontendPort() +
-												':' + this.zeroMqProxy.getBackendPort()))
+										setConnectUrl("tcp://localhost:" + zeroMqProxyToUse.getFrontendPort() +
+												':' + zeroMqProxyToUse.getBackendPort()))
 								.doOnError((error) ->
 										logger.error(error,
-												() -> "The provided '" + this.zeroMqProxy + "' has not been started"));
+												() -> "The provided '" + zeroMqProxyToUse + "' has not been started"));
 					}
 					else {
 						return Mono.empty();

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/channel/package-info.java

@@ -1,5 +1,5 @@
 /**
  * Provides classes for message channels support over ZeroMQ.
  */
-@org.springframework.lang.NonNullApi
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.zeromq.channel;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/dsl/package-info.java

@@ -1,6 +1,5 @@
 /**
  * Provides classes for supporting ZeroMQ component via Java DSL.
  */
-@org.springframework.lang.NonNullApi
-@org.springframework.lang.NonNullFields
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.zeromq.dsl;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/ZeroMqMessageProducer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -24,6 +24,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 
+import org.jspecify.annotations.Nullable;
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZFrame;
@@ -43,7 +44,6 @@
 import org.springframework.integration.zeromq.ZeroMqUtils;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
@@ -77,6 +77,7 @@ public class ZeroMqMessageProducer extends MessageProducerSupport {
 
 	private final SocketType socketType;
 
+	@SuppressWarnings("NullAway.Init")
 	private InboundMessageMapper<byte[]> messageMapper;
 
 	private Consumer<ZMQ.Socket> socketConfigurer = (socket) -> {
@@ -91,6 +92,7 @@ public class ZeroMqMessageProducer extends MessageProducerSupport {
 	@Nullable
 	private String connectUrl;
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile Mono<ZMQ.Socket> socketMono;
 
 	private volatile boolean unwrapTopic = true;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/inbound/package-info.java

@@ -1,5 +1,5 @@
 /**
  * Provides classes for inbound channel adapters over ZeroMQ.
  */
-@org.springframework.lang.NonNullApi
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.zeromq.inbound;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/ZeroMqMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2024 the original author or authors.
+ * Copyright 2020-2025 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.
@@ -22,6 +22,7 @@
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import org.jspecify.annotations.Nullable;
 import org.zeromq.SocketType;
 import org.zeromq.ZContext;
 import org.zeromq.ZFrame;
@@ -45,7 +46,6 @@
 import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
 import org.springframework.integration.support.management.ManageableLifecycle;
 import org.springframework.integration.zeromq.ZeroMqUtils;
-import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
@@ -77,19 +77,23 @@ public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler
 
 	private final Scheduler publisherScheduler = Schedulers.newSingle("zeroMqMessageHandlerScheduler");
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile Mono<ZMQ.Socket> socketMono;
 
+	@SuppressWarnings("NullAway.Init")
 	private OutboundMessageMapper<byte[]> messageMapper;
 
 	private Consumer<ZMQ.Socket> socketConfigurer = (socket) -> {
 	};
 
 	private Expression topicExpression = new SupplierExpression<>(() -> null);
 
+	@SuppressWarnings("NullAway.Init")
 	private EvaluationContext evaluationContext;
 
 	private volatile boolean initialized;
 
+	@SuppressWarnings("NullAway.Init")
 	private volatile Disposable socketMonoSubscriber;
 
 	private volatile boolean wrapTopic = true;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/outbound/package-info.java

@@ -1,5 +1,5 @@
 /**
  * Provides classes for outbound channel adapters over ZeroMQ.
  */
-@org.springframework.lang.NonNullApi
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.zeromq.outbound;

spring-integration-zeromq/src/main/java/org/springframework/integration/zeromq/package-info.java

@@ -1,5 +1,5 @@
 /**
  * Provides common classes for supporting ZeroMQ components.
  */
-@org.springframework.lang.NonNullApi
+@org.jspecify.annotations.NullMarked
 package org.springframework.integration.zeromq;