Add `spring-integration-netty` module

[clayly]

In what version(s) of Spring Integration are you seeing this issue?

6.1.1

Describe the bug

There is no explanation in documentation about how to use NIO for UDP neither corresponding methods or attributes in API.

To Reproduce

Look here: https://docs.spring.io/spring-integration/reference/html/ip.html#ip

Expected behavior

There should be explanation how to use NIO for UDP or clear explatation why it's impossible.

Sample

There is no examles of using UDP adapters with NIO.

[garyrussell]

There is currently no support for NIO with UDP in this project.

It would only make sense to implement it if a single inbound adapter could support reading from multiple sockets, which is currently not the case. It would require an architectural change.

See https://stackoverflow.com/questions/16048284/whats-the-point-of-using-udp-with-nio for a discussion about this.

Contributions are welcome.

[clayly]

It would only make sense to implement it if a single inbound adapter could support reading from multiple sockets, which is currently not the case. It would require an architectural change.

I think it would also make sense to implement it even if there is multiple inbound adapters with single socket in each but they all share same EventLoopGroup (e. g. for Netty), which is quite normal. What do you think?

[garyrussell]

Right, but that's still an architectural change, to externalize the event handling to a shared component.

[artembilan]

May be we just need to look into a new pair of channel adapters based on Reactor Netty:

https://projectreactor.io/docs/netty/release/reference/index.html#udp-server
https://projectreactor.io/docs/netty/release/reference/index.html#udp-client
?

[clayly]

May be we just need to look into a new pair of channel adapters based on Reactor Netty:

https://projectreactor.io/docs/netty/release/reference/index.html#udp-server https://projectreactor.io/docs/netty/release/reference/index.html#udp-client ?

It's what I use now instead of UDP integration adapters.

[artembilan]

I would suggest to repurpose this issue into something like new spring-integration-netty module.
And there we would look into a UDP part how to implement it based on the Netty library.
Or something what is on top of that, like Reactor Netty: https://projectreactor.io/docs/netty/milestone/reference/about-doc.html

[goutamadwant]

@artembilan I reviewed the current Spring Integration, the existing UDP implementation and the current Reactor Netty UDP APIs.

As of today this is what we have :

• There is no spring-integration-netty module.
• There are no Spring Integration UDP adapters based directly on Netty or Reactor Netty.
• The existing UDP adapters in spring-integration-ip are still based on java.net.DatagramSocket and DatagramPacket.
• The inbound UnicastReceivingChannelAdapter performs blocking DatagramSocket.receive() calls on a dedicated task-executor thread.
• The existing outbound UnicastSendingMessageHandler also uses blocking DatagramSocket.send().
• Reactor Netty already provides UdpServer, UdpClient, shared event-loop resources, native transports, reactive lifecycle management, UDP metrics, and multicast operations.

I think fhe issue tjat was originally reported was about documenting UDP NIO support, but the current direction as per @artembilan is broader: add a new spring-integration-netty module, starting with UDP adapters.

That is a better design than adding an nio flag to the existing UDP adapters. The lifecycle, configuration, error handling, threading, buffer ownership, and reactive send semantics are sufficiently different that trying to hide both implementations behind the current classes would make them harder to maintain.

Recommended scope

I suggest treating the first contribution as an initial UDP-focused milestone for the new module.

The first pull request should provide:

• A new spring-integration-netty module
• Reactor Netty UDP inbound channel adapter
• Reactor Netty UDP outbound reactive message handler
• Java DSL support
• Shared/default Reactor Netty event-loop support
• Message conversion and existing IP header compatibility
• Deterministic lifecycle and resource cleanup
• Unicast support
• Documentation and tests

The first version should not attempt to replace all existing spring-integration-ip functionality.

The following features should be explicitly deferred unless maintainers want them in the initial scope:

• TCP adapters and gateways
• UDP request/reply gateways
• Application-level UDP acknowledgments
• The existing four-byte lengthCheck protocol
• Multicast support
• Dynamic per-message destinations
• Unix domain datagram sockets
• XML namespace support
• Automatic reconnect/rebind policies

These can be introduced incrementally after the basic lifecycle and reactive behavior are established.

Module structure

Proposed module:

spring-integration-netty
└── src/main/java/org/springframework/integration/netty
    ├── udp
    │   ├── inbound
    │   ├── outbound
    │   └── support
    └── dsl

Suggested initial classes:

org.springframework.integration.netty.udp.inbound.NettyUdpInboundChannelAdapter
org.springframework.integration.netty.udp.outbound.NettyUdpOutboundMessageHandler
org.springframework.integration.netty.udp.support.NettyUdpMessageMapper
org.springframework.integration.netty.dsl.NettyUdp
org.springframework.integration.netty.dsl.NettyUdpInboundChannelAdapterSpec
org.springframework.integration.netty.dsl.NettyUdpOutboundMessageHandlerSpec

The module should depend on:

project('spring-integration-netty') {
    description = 'Spring Integration Netty Support'

    dependencies {
        api project(':spring-integration-ip')
        api 'io.projectreactor.netty:reactor-netty-core'
    }
}

Depending on spring-integration-ip allows the Netty implementation to preserve existing IpHeaders names and, where appropriate, wire-level compatibility with the current UDP adapters.

The Reactor Netty version should continue to be managed through the Reactor BOM already used by the project.

The new module will be included automatically by the current settings.gradle module discovery, but it should also be covered by the generated BOM, aggregated API documentation, distribution, reference navigation, and “What’s New” documentation.

Inbound adapter

The inbound component should extend MessageProducerSupport:

public final class NettyUdpInboundChannelAdapter
        extends MessageProducerSupport {
}

Its primary constructor should accept a configured Reactor Netty UdpServer:

public NettyUdpInboundChannelAdapter(UdpServer udpServer) {
    // ...
}

Accepting UdpServer rather than recreating every Reactor Netty option as a Spring Integration property gives applications access to:

• Host and port
• LoopResources
• Native transport selection
• Netty ChannelOption values
• Wire logging
• Reactor Netty metrics
• Channel pipeline customization
• Bind address customization
• IPv4/IPv6 configuration

For example:

LoopResources udpLoops =
        LoopResources.create("spring-integration-udp", 2, true);

UdpServer server =
        UdpServer.create()
                .host("0.0.0.0")
                .port(9000)
                .runOn(udpLoops);

Multiple inbound and outbound adapters can use the same LoopResources, directly addressing the original request to share event-processing resources between multiple UDP sockets.

Reactor Netty also uses shared UDP event-loop resources by default, so applications that do not need custom isolation should not be required to create their own group.

Inbound lifecycle

doStart() should:

1. Attach the Spring Integration receive pipeline to the configured UdpServer.
2. Bind with a configurable startup timeout.
3. Store the resulting Reactor Netty Connection.
4. Record the actual bound address and port.
5. Publish a listening event only after binding succeeds.
6. Mark the adapter as listening/running.

doStop() should:

1. Cancel the Spring Integration publisher subscription.
2. Dispose the bound Reactor Netty connection.
3. Wait up to a configurable shutdown timeout.
4. Clear connection and bound-address state.
5. Allow the same endpoint to be started again.

The adapter must own and dispose its bound Connection, but it must not dispose shared or externally supplied LoopResources. Disposing shared loop resources from one adapter would stop all other clients and servers using them.

The adapter should expose the actual bound port so that port 0 can be used in tests and dynamic deployments:

public int getBoundPort();

Inbound packet processing

The Reactor Netty receive pipeline should consume io.netty.channel.socket.DatagramPacket instances from UdpInbound.receiveObject().

A packet must not be passed directly to an arbitrary Spring Integration flow because its ByteBuf is reference-counted and normally belongs to the Netty pipeline.

The safe sequence is:

1. Read the sender and recipient addresses.
2. Copy the readable bytes from the packet’s ByteBuf into a byte[].
3. Build the Spring message and IP headers.
4. Allow Reactor Netty to release the original packet.
5. Dispatch the Spring message outside the Netty event-loop thread when the downstream channel may block.

The default produced message should contain a byte[] payload and preserve the existing headers:

IpHeaders.HOSTNAME
IpHeaders.IP_ADDRESS
IpHeaders.PORT
IpHeaders.PACKET_ADDRESS
IpHeaders.LOCAL_ADDRESS

Reverse DNS lookup should remain disabled by default, matching the current UDP adapter behavior.

A configurable InboundMessageMapper<byte[]> or MessageConverter should be supported. The mapper should receive the address metadata as headers rather than receiving the Netty packet itself.

This keeps Netty reference-counted objects out of the default user-facing API and makes custom conversion safer.

Event-loop safety

The adapter must not call an arbitrary synchronous downstream flow directly on a Netty event-loop thread.

A DirectChannel subscriber can perform blocking database, file, HTTP, or other application work. Running that work on the UDP event loop would stall every socket sharing the loop.

The inbound component should therefore support a configurable dispatch Scheduler or Executor.

A possible API is:

public void setDispatchScheduler(Scheduler dispatchScheduler);

The receive pipeline can copy the datagram on the Netty event loop and then use publishOn(dispatchScheduler) before calling the Spring Integration message producer.

The default should be chosen with safety in mind, such as Reactor’s shared bounded-elastic scheduler, while allowing fully reactive applications to provide a more appropriate scheduler.

The implementation should also expose or internally define a bounded prefetch value. Reactor backpressure can pause local reads, but it cannot make UDP reliable. When the consumer is slower than the sender, the operating-system receive buffer can overflow and datagrams can still be dropped.

The documentation must state this explicitly and must not claim that Reactive Streams backpressure guarantees UDP delivery.

Error handling

An exception while mapping or dispatching one datagram should not automatically terminate the entire UDP server.

Per-packet failures should be converted to a MessagingException and sent to the configured Spring Integration errorChannel using the facilities already provided by MessageProducerSupport.

Errors that make the underlying channel unusable should terminate the connection and update the adapter lifecycle state.

The design should distinguish:

• Packet conversion errors
• Downstream message delivery errors
• Bind failures
• Netty channel failures
• Endpoint cancellation
• Normal shutdown

Logging must include the relevant local/remote addresses but should not dump datagram contents unless Reactor Netty wire logging was explicitly enabled.

Outbound message handler

The outbound component should be reactive:

public final class NettyUdpOutboundMessageHandler
        extends AbstractReactiveMessageHandler
        implements ManageableLifecycle {
}

Using AbstractReactiveMessageHandler is important. Blocking on a Reactor Netty send would defeat the purpose of the module and could deadlock when invoked from another reactive flow.

The primary constructor should accept a configured UdpClient:

public NettyUdpOutboundMessageHandler(UdpClient udpClient) {
    // ...
}

For the initial implementation, the UdpClient should represent one fixed remote destination.

The handler should:

1. Connect and create a reusable UDP channel during start().
2. Store the resulting Connection.
3. Convert each request message to byte[].
4. Wrap the bytes in a Netty ByteBuf.
5. Send the datagram through the cached connection.
6. Return the Reactor send completion as Mono<Void>.
7. Propagate send failures through the normal Spring Integration reactive error path.
8. Dispose the connection during stop() and destroy().

The send operation should look conceptually similar to:

@Override
protected Mono<Void> handleMessageInternal(Message<?> message) {
    return Mono.defer(() -> {
        Connection connection = getRequiredConnection();
        byte[] payload = this.messageMapper.fromMessage(message);

        return connection.outbound()
                .sendObject(Mono.fromSupplier(
                        () -> Unpooled.wrappedBuffer(payload)))
                .then();
    });
}

The exact implementation should ensure that any allocated ByteBuf is released on pre-send failures and that Netty owns the buffer after a successful write submission.

The first version should fail clearly when a message arrives before the endpoint has started or after the connection has been disposed.

Automatic reconnection should be discussed separately rather than silently retrying UDP sends.

Message mapping

The default outbound mapper should support at least:

• byte[]
• String, using UTF-8 by default

A configurable OutboundMessageMapper<byte[]> or MessageConverter should support other payload types.

Keeping the mapper based on byte[], rather than exposing Netty ByteBuf, has several advantages:

• It follows patterns already used by reactive Spring Integration modules such as ZeroMQ.
• It prevents accidental use-after-release errors.
• It keeps application code independent of Netty buffer ownership.
• It makes unit testing easier.
• It allows the handler to own creation and release of outbound Netty buffers.

A raw ByteBuf mode could be considered later, but it would require a precise ownership contract.

Dynamic destinations

The existing UDP outbound adapter supports a destination expression. That should not automatically be copied into the first Netty implementation.

A connected Reactor Netty UdpClient naturally represents one remote destination. Sending to a different recipient for every message may require an unconnected bound datagram channel or a managed connection/channel cache.

That should be a follow-up design decision.

If dynamic destinations are added, the existing IpHeaders.PACKET_ADDRESS header and destination expressions should be supported, and the implementation must avoid creating a new UDP channel for every message.

Java DSL

A focused NettyUdp DSL factory would avoid a broad Netty class before the module supports additional protocols.

Suggested API:

public final class NettyUdp {

    public static NettyUdpInboundChannelAdapterSpec inboundAdapter(
            UdpServer udpServer);

    public static NettyUdpInboundChannelAdapterSpec inboundAdapter(
            String host, int port);

    public static NettyUdpOutboundMessageHandlerSpec outboundAdapter(
            UdpClient udpClient);

    public static NettyUdpOutboundMessageHandlerSpec outboundAdapter(
            String host, int port);
}

Example:

@Bean
LoopResources udpLoopResources() {
    return LoopResources.create("integration-udp", 2, true);
}

@Bean
IntegrationFlow udpInboundFlow(LoopResources udpLoopResources) {
    UdpServer server =
            UdpServer.create()
                    .host("0.0.0.0")
                    .port(9000)
                    .runOn(udpLoopResources);

    return IntegrationFlow
            .from(NettyUdp.inboundAdapter(server))
            .handle(message -> {
                byte[] payload = (byte[]) message.getPayload();
                // Process the datagram.
            })
            .get();
}

@Bean
IntegrationFlow udpOutboundFlow(LoopResources udpLoopResources) {
    UdpClient client =
            UdpClient.create()
                    .host("127.0.0.1")
                    .port(9000)
                    .runOn(udpLoopResources);

    return IntegrationFlow
            .from("udpOutboundRequests")
            .handle(NettyUdp.outboundAdapter(client))
            .get();
}

Lifecycle ownership must be documented in this example: the endpoint owns its Connection, while the application owns the shared LoopResources.

Following newer modules such as spring-integration-grpc, Java DSL support should be sufficient for the first contribution. XML namespace support can be considered later if there is demand.

Existing UDP protocol compatibility

The current DatagramPacketMessageMapper supports two Spring Integration-specific wire features:

• Four-byte length prefix
• Application-level acknowledgment headers and responses

The basic Netty adapter should use ordinary UDP datagrams and remain interoperable with any UDP application.

If the legacy Spring Integration framing options are later added, the encoding and decoding logic should be extracted into a transport-independent byte-array codec and shared by both the java.net and Netty adapters.

The new implementation should not duplicate the acknowledgment parsing and length-prefix protocol in a second mapper.

Application-level acknowledgment support is more involved because it requires the inbound Netty channel to send a response to the packet sender and the outbound component to correlate acknowledgments asynchronously. That should be a separate follow-up after basic unicast send/receive behavior is stable.

Multicast

Reactor Netty’s UdpInbound supports joining and leaving multicast groups, so multicast is technically feasible.

However, multicast adds lifecycle and configuration concerns:

• Network interface selection
• Join and leave timing
• Multiple groups
• IPv4 versus IPv6
• Loopback mode
• TTL
• Platform-dependent test behavior

I recommend adding multicast in a second pull request rather than combining it with the initial module and unicast adapter implementation.

Observability and metrics

The Spring Integration components should retain the existing endpoint observation behavior inherited from MessageProducerSupport and AbstractReactiveMessageHandler.

Reactor Netty transport metrics should remain opt-in through the supplied UdpServer and UdpClient.

The documentation should warn that Reactor Netty UDP metrics can include remote.address tags. Applications receiving packets from many addresses should configure a Micrometer tag limit to avoid unbounded metric cardinality.

A separate Spring Integration metric implementation should not duplicate Reactor Netty’s byte and error metrics in the first version.

Testing plan

Module and dependency tests

• spring-integration-netty is included in the build and BOM.
• reactor-netty-core is resolved through the existing Reactor BOM.
• No Netty dependency is added to applications using only spring-integration-ip.

Inbound tests

• Bind to a configured host and port.
• Bind to port 0 and expose the actual port.
• Receive a datagram as byte[].
• Populate all expected IpHeaders.
• Preserve sender address and port.
• Disable reverse DNS lookup by default.
• Use a custom inbound mapper.
• Route mapping failures to the configured error channel.
• Start, stop, and restart the same endpoint.
• Release every inbound ByteBuf.
• Do not perform downstream blocking work on the Netty event-loop thread.

Outbound tests

• Reuse one connected UDP channel for multiple messages.
• Send byte[] payloads.
• Send String payloads with UTF-8 conversion.
• Use a custom outbound mapper.
• Return completion only after the Netty write completes.
• Propagate connection and write failures.
• Reject sends while stopped.
• Start, stop, and restart.
• Release buffers on errors and cancellation.

Shared event-loop tests

• Run multiple inbound adapters and outbound handlers on one LoopResources.
• Verify that stopping one endpoint does not stop the shared event loops.
• Verify that the remaining endpoints continue to send and receive.
• Verify that the module does not dispose application-owned LoopResources.

End-to-end tests

• Netty outbound to Netty inbound.
• Existing spring-integration-ip outbound to Netty inbound.
• Netty outbound to existing spring-integration-ip inbound.
• Large datagram near the configured receive-buffer limit.
• Empty datagram behavior.
• IPv4 loopback.
• IPv6 loopback when available.
• Burst traffic and documented drop behavior under a deliberately slow consumer.

Buffer leak tests

Run relevant tests with Netty leak detection enabled:

-Dio.netty.leakDetection.level=paranoid

The tests should cover success, mapping failure, downstream failure, cancellation, and shutdown paths.

Lifecycle tests

• Bind failure is reported synchronously during startup.
• Startup timeout is honored.
• Shutdown timeout is honored.
• Unexpected connection disposal updates endpoint state.
• No messages are emitted after stop completes.
• Restart creates a new connection without reusing a disposed channel.

Documentation

Add a new reference section describing:

• Dependency coordinates
• Why this module is separate from spring-integration-ip
• Reactor Netty versus the existing blocking UDP adapters
• Inbound and outbound Java DSL examples
• Shared LoopResources
• Event-loop ownership and shutdown
• Supported payload types
• IP headers
• Error-channel behavior
• UDP delivery limitations
• Backpressure limitations
• Native transport selection
• Reactor Netty metrics and tag cardinality
• Features not yet supported
• Migration guidance for applications currently building Reactor Netty UDP flows manually

The existing IP documentation should also state clearly that the spring-integration-ip UDP adapters use java.net.DatagramSocket, while non-blocking Reactor Netty UDP support is provided by the new module.

Recommended implementation sequence

1. Confirm that the first spring-integration-netty milestone is UDP-only.
2. Add the new module and reactor-netty-core dependency.
3. Implement the inbound adapter lifecycle and safe datagram-to-message conversion.
4. Add dispatch scheduling so application work does not block Netty event loops.
5. Implement the reactive outbound handler with a reusable UDP connection.
6. Add inbound and outbound message-mapper customization.
7. Preserve existing IpHeaders.
8. Add the NettyUdp Java DSL and specs.
9. Add lifecycle, interoperability, shared-loop, and leak-detection tests.
10. Add reference documentation and “What’s New” notes.
11. Address multicast, dynamic destinations, length checking, and acknowledgments in focused follow-up issues.

This design provides the requested non-blocking UDP support and shared event-loop architecture without changing the behavior or dependency footprint of the existing spring-integration-ip adapters. Let me know your thoughts on this @artembilan