CAMEL-26179: camel-milo - a milo-client write of a null body fails the exchange

catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/milo-client-component.adoc

@@ -171,6 +171,9 @@ and `${body.value.value}` is the value. A producer writes the message body to it
               allowedSecurityPolicies: None
 ----
 
+A producer does not write a `null` body: the exchange fails with an `InvalidPayloadException`. To write a null
+value, set the body to `Variant.NULL_VALUE` or to a `DataValue` that carries it.
+
 === Monitoring several nodes
 
 A consumer can monitor several nodes with the `nodes` option instead of `node` (the two options cannot be combined).

components/camel-milo/src/main/docs/milo-client-component.adoc

@@ -171,6 +171,9 @@ and `${body.value.value}` is the value. A producer writes the message body to it
               allowedSecurityPolicies: None
 ----
 
+A producer does not write a `null` body: the exchange fails with an `InvalidPayloadException`. To write a null
+value, set the body to `Variant.NULL_VALUE` or to a `DataValue` that carries it.
+
 === Monitoring several nodes
 
 A consumer can monitor several nodes with the `nodes` option instead of `node` (the two options cannot be combined).

components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientProducer.java

@@ -22,6 +22,7 @@
 
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
 import org.apache.camel.Message;
 import org.apache.camel.component.milo.MiloConstants;
 import org.apache.camel.support.DefaultAsyncProducer;
@@ -71,10 +72,19 @@ protected void doStop() throws Exception {
     public boolean process(Exchange exchange, AsyncCallback async) {
         final Message msg = exchange.getIn();
         final Object value = msg.getBody();
+        final boolean read = msg.getHeaders().containsKey(HEADER_NODE_IDS);
+
+        if (value == null && !read && this.methodId == null) {
+            // a write needs a value, a null body is not written
+            exchange.setException(new InvalidPayloadException(
+                    exchange, Object.class, msg, "set the body to Variant.NULL_VALUE to write a null value"));
+            async.done(true);
+            return true;
+        }
 
         final CompletableFuture<?> future;
 
-        if (msg.getHeaders().containsKey(HEADER_NODE_IDS)) {
+        if (read) {
             final List<String> nodeIds = msg.getHeader(HEADER_NODE_IDS, List.class);
             final List<ExpandedNodeId> expandedNodeIds
                     = nodeIds.stream().map(String.class::cast).map(ExpandedNodeId::parse).collect(Collectors.toList());

components/camel-milo/src/test/java/org/apache/camel/component/milo/ProducerOnSharedConnectionTest.java

@@ -46,6 +46,7 @@
 import org.eclipse.milo.opcua.sdk.server.AddressSpace;
 import org.eclipse.milo.opcua.stack.core.StatusCodes;
 import org.eclipse.milo.opcua.stack.core.UaException;
+import org.eclipse.milo.opcua.stack.core.UaSerializationException;
 import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
 import org.eclipse.milo.opcua.stack.core.types.builtin.ExpandedNodeId;
 import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
@@ -65,6 +66,7 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -265,6 +267,57 @@ void writeThatTimesOutKeepsTheConnection() throws Exception {
         }
     }
 
+    /**
+     * A write of a body that the client cannot encode fails the exchange. It does not rebuild the connection: the
+     * consumer of another node reports no loss and keeps receiving values in the same session.
+     */
+    @Test
+    @Timeout(value = 60, unit = TimeUnit.SECONDS)
+    void writeOfABodyThatCannotBeEncodedKeepsTheConnection() throws Exception {
+        template.sendBody("milo-server:myitem1", "Foo");
+        template.sendBody("milo-server:myitem2", "Foo");
+        item2Endpoint.expectedMessageCount(1);
+        testBody(item2Endpoint.message(0), assertGoodValue("Foo"));
+        context.getRouteController().startRoute("item2");
+        item2Endpoint.assertIsSatisfied();
+
+        final MiloClientEndpoint endpoint = context.getEndpoint(resolve(MILO_CLIENT_BASE + ITEM_1), MiloClientEndpoint.class);
+        final MiloClientConnection connection = endpoint.createConnection();
+        try {
+            final OpcUaClient client = connection.getOpcUaClient();
+            final NodeId sessionId = client.getSession().getSessionId();
+
+            // the consumer receives the next value only, no loss report and not its value again
+            item2Endpoint.reset();
+            item2Endpoint.expectedMessageCount(1);
+            item2Endpoint.setAssertPeriod(2_000);
+            testBody(item2Endpoint.message(0), assertGoodValue("Bar"));
+
+            final Exchange written = template.send(resolve(MILO_CLIENT_BASE + ITEM_1), exchange -> {
+                exchange.getMessage().setBody(Map.of("key", "value"));
+                exchange.getMessage().setHeader(MiloConstants.HEADER_AWAIT, true);
+            });
+            assertNotNull(ObjectHelper.getException(UaSerializationException.class, written.getException()),
+                    () -> "Expected a UaSerializationException, but the exchange failed with " + written.getException());
+
+            // a write on the same connection still succeeds
+            final Exchange rewritten = template.send(resolve(MILO_CLIENT_BASE + ITEM_1), exchange -> {
+                exchange.getMessage().setBody(new Variant("Bar"));
+                exchange.getMessage().setHeader(MiloConstants.HEADER_AWAIT, true);
+            });
+            assertNull(rewritten.getException());
+            assertEquals(StatusCode.GOOD, rewritten.getMessage().getBody());
+
+            template.sendBody("milo-server:myitem2", "Bar");
+            item2Endpoint.assertIsSatisfied();
+
+            assertSame(client, connection.getOpcUaClient());
+            assertEquals(sessionId, client.getSession().getSessionId());
+        } finally {
+            endpoint.releaseConnection(connection);
+        }
+    }
+
     /**
      * A write in flight on a connection that Camel replaces fails with the status of its closed connection. Its failure
      * does not rebuild the connection that replaced it.

components/camel-milo/src/test/java/org/apache/camel/component/milo/WriteClientTest.java

@@ -17,6 +17,8 @@
 package org.apache.camel.component.milo;
 
 import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadException;
 import org.apache.camel.Produce;
 import org.apache.camel.ProducerTemplate;
 import org.apache.camel.RoutesBuilder;
@@ -27,10 +29,13 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import static org.apache.camel.component.milo.NodeIds.nodeValue;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 
 /**
  * Unit tests for writing from the client side
@@ -169,6 +174,25 @@ public void testWrite3() throws Exception {
         MockEndpoint.assertIsSatisfied(context);
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = { true, false })
+    public void testWriteNullBodyFails(final boolean await) throws Exception {
+        // item 1 receives the next write only
+        this.test1Endpoint.setExpectedCount(1);
+        testBody(this.test1Endpoint.message(0), assertGoodValue("Foo1"));
+
+        // send
+        final Exchange exchange = this.producer1.send(e -> {
+            e.getMessage().setBody(null);
+            e.getMessage().setHeader("CamelMiloAwait", await);
+        });
+        sendValue(this.producer1, new Variant("Foo1"));
+
+        // assert
+        assertInstanceOf(InvalidPayloadException.class, exchange.getException());
+        MockEndpoint.assertIsSatisfied(context);
+    }
+
     private static void sendValue(final ProducerTemplate producerTemplate, final Variant variant) {
         // we always write synchronously since we do need the message order
         producerTemplate.sendBodyAndHeader(variant, "CamelMiloAwait", true);

docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc

@@ -2583,6 +2583,14 @@ A `sessionTimeout` configured on the `milo-client` or `milo-browse` component (f
 `camel.component.milo-client.sessionTimeout`) did not reach its endpoints, so their sessions used the default of Milo,
 120000 milliseconds. The endpoints now use it, unless they set `sessionTimeout` themselves.
 
+=== camel-milo - a write of a null body fails the exchange
+
+A `milo-client` producer no longer writes a `null` body; the exchange now fails with an `InvalidPayloadException`,
+whether or not the write is awaited. Before, it wrote a null value, which a server typically refused with
+`Bad_TypeMismatch`, and an awaited write completed without an error and with that status as the body. To write a null
+value, set the body to `Variant.NULL_VALUE` or to a `DataValue` that carries it. A method call with a `null` body still
+calls the method without input arguments.
+
 === camel-infinispan - the key option of the producer takes precedence over the key header
 
 The `infinispan` and `infinispan-embedded` producers now use the `key` option of the endpoint when it is set, and use the