Java Client SDK can't get response after sending live message to Device

[yuguanxiang]

my environment:

java client sdk ---- Ditto ----- MQTT broker ------ Device

my problem:

i used java sdk to send live message API to device, and i saw Device received the ditto protocol message, and then Device modified the message content

• add json key “status” 201
• modify path “inbox” to “outbox”,
• modify header key ditto-message-direction “TO” to “FROM”,
  then send the message back as response, i observe the message from mqtt, i expect sdk API would get response, but it got no response.

mqtt connection configuration

 {
    "connectivity": {
        "1": {
            "type": "connectivity.responses:retrieveConnection",
            "status": 200,
            "connection": {
                "id": "mqtt-example-connection-123",
                "name": null,
                "connectionType": "mqtt",
                "connectionStatus": "open",
                "uri": "tcp://10.2.177.40:1883",
                "sources": [
                    {
                        "addresses": [
                            "my.test/#"
                        ],
                        "consumerCount": 1,
                        "qos": 0,
                        "authorizationContext": [
                            "nginx:ditto"
                        ],
                        "headerMapping": {},
                        "replyTarget": {
                            "address": "{{header:reply-to}}",
                            "headerMapping": {},
                            "expectedResponseTypes": [
                                "response",
                                "error"
                            ],
                            "enabled": true
                        }
                    }
                ],
                "targets": [
                    {
                        "address": "my.test/{{ thing:id }}",
                        "topics": [
                            "_/_/things/twin/events",
                            "_/_/things/live/messages"
                        ],
                        "qos": 0,
                        "authorizationContext": [
                            "nginx:ditto"
                        ],
                        "headerMapping": {}
                    }
                ],
                "clientCount": 1,
                "failoverEnabled": true,
                "validateCertificates": true,
                "processorPoolSize": 5,
                "tags": []
            }
        }
    }
}

My SDK API demo codes

the output print codes was not executed, as i did not see the print info

         final DittoClient client = DittoClients.newInstance(createMessagingProvider())
                .connect()
                .toCompletableFuture()
                .join();
        client.live().forId(ThingId.of("my.test", "thing3"))
                .message()
                .to()
                .subject("TO--hello.world")
                .payload("I am a TeapotX")
                .send(String.class, (response, throwable) -> {
                    System.out.println("XXXX: Got response: " + response.getPayload().orElse(null));  //<-----------Here, no any output is seen
                        }
                );

Message from MQTT (mosquitto)

message1(Device received)

{
        "topic":        "my.test/thing3/things/live/messages/TO--hello.world",
        "headers":      {
                "version":      2,
                "response-required":    true,
                "ditto-originator":     "nginx:ditto",
                "requested-acks":       [],
                "correlation-id":       "3711d8da-7677-491b-bffc-592f8a628b37",
                "content-type": "text/plain",
                "ditto-message-thing-id":       "my.test:thing3",
                "ditto-message-subject":        "TO--hello.world",
                "ditto-message-direction":      "TO"
        },
        "value":        "I am a TeapotX",
        "path": "/inbox/messages/TO--hello.world"
}

message2(Device sent as response)

{
        "topic":        "my.test/thing3/things/live/messages/TO--hello.world",
        "headers":      {
                "version":      2,
                "response-required":    true,
                "ditto-originator":     "nginx:ditto",
                "requested-acks":       [],
                "correlation-id":       "3711d8da-7677-491b-bffc-592f8a628b37",
                "content-type": "text/plain",
                "ditto-message-thing-id":       "my.test:thing3",
                "ditto-message-subject":        "TO--hello.world",
                "ditto-message-direction":      "FROM"
        },
        "value":        "I am a TeapotX",
        "status":       200,
        "path": "/outbox/messages/TO--hello.world"
}

some logs (from ditto-gateway)

dead letters encountered

~ $ cat /var/log/ditto/gateway.log|grep 3711d8da
{"@timestamp":"2021-11-24T07:49:48.964+01:00","@version":"1","message":"Forwarding signal with ID <my.test:thing3> and type <messages.commands:thingMessage> to concierge enforcer","logger_name":"org.eclipse.ditto.concierge.api.actors.ConciergeForwarderActor","thread_name":"ditto-cluster-akka.actor.default-dispatcher-19","level":"INFO","level_value":20000,"akkaAddress":"akka://ditto-cluster@172.18.0.8:2551","sourceThread":"ditto-cluster-akka.actor.default-dispatcher-16","akkaSource":"akka://ditto-cluster/user/gatewayRoot/conciergeForwarder","sourceActorSystem":"ditto-cluster","x-correlation-id":"3711d8da-7677-491b-bffc-592f8a628b37","akkaTimestamp":"06:49:48.964UTC","appname":"gateway","instance_index":"1"}
{"@timestamp":"2021-11-24T07:49:48.966+01:00","@version":"1","message":"Starting to wait for all requested acknowledgements <[live-response]> for a maximum duration of <PT1M5S>.","logger_name":"org.eclipse.ditto.internal.models.acks.AcknowledgementAggregatorActor","thread_name":"ditto-cluster-akka.actor.default-dispatcher-19","level":"INFO","level_value":20000,"akkaAddress":"akka://ditto-cluster@172.18.0.8:2551","sourceThread":"ditto-cluster-akka.actor.default-dispatcher-18","akkaSource":"akka://ditto-cluster/user/gatewayRoot/streaming/12-14f245a2-a593-44da-a80c-1f4688b74b62/ackr0-3711d8da-7677-491b-bffc-592f8a628b37","sourceActorSystem":"ditto-cluster","x-correlation-id":"3711d8da-7677-491b-bffc-592f8a628b37","akkaTimestamp":"06:49:48.965UTC","appname":"gateway","instance_index":"1"}
{"@timestamp":"2021-11-24T07:49:49.045+01:00","@version":"1","message":"Message [org.eclipse.ditto.messages.model.signals.commands.SendThingMessageResponse] from Actor[akka://ditto-cluster@172.18.0.9:2551/deadLetters] to Actor[akka://ditto-cluster/user/gatewayRoot/streaming/12-14f245a2-a593-44da-a80c-1f4688b74b62/ackr0-3711d8da-7677-491b-bffc-592f8a628b37#-1315189962] was not delivered. [10] dead letters encountered, no more dead letters will be logged in next [5.000 min]. If this is not an expected behavior then Actor[akka://ditto-cluster/user/gatewayRoot/streaming/12-14f245a2-a593-44da-a80c-1f4688b74b62/ackr0-3711d8da-7677-491b-bffc-592f8a628b37#-1315189962] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.","logger_name":"akka.actor.EmptyLocalActorRef","thread_name":"ditto-cluster-akka.actor.default-dispatcher-19","level":"INFO","level_value":20000,"akkaAddress":"akka://ditto-cluster@172.18.0.8:2551","sourceThread":"ditto-cluster-akka.actor.default-dispatcher-16","akkaSource":"akka://ditto-cluster/user/gatewayRoot/streaming/12-14f245a2-a593-44da-a80c-1f4688b74b62/ackr0-3711d8da-7677-491b-bffc-592f8a628b37","sourceActorSystem":"ditto-cluster","akkaMessageClass":"org.eclipse.ditto.messages.model.signals.commands.SendThingMessageResponse","akkaTimestamp":"06:49:49.045UTC","tags":["akkaDeadLetter"],"appname":"gateway","instance_index":"1"}
~ $

So here, i have some questions:

1. is the ditto connection config is correct?
2. does Device correctly modify the request message ?
3. i'm not sure whether my client sdk call is correct, and which is right way?
4. how to set the timeout for client sdk
5. what topic should Device publish message to?

thanks a lot!

[thjaeckle]

@yuguanxiang could you please try to send the following response instead (all other fields are not necessary and may even disturb the processing - just to rule that possibility out):

{
        "topic":        "my.test/thing3/things/live/messages/TO--hello.world",
        "headers":      {
                "version":      2,
                "correlation-id":       "3711d8da-7677-491b-bffc-592f8a628b37",
                "content-type": "text/plain"
        },
        "value":        "I am a TeapotX",
        "status":       200,
        "path": "/outbox/messages/TO--hello.world"
}

[yuguanxiang]

@yuguanxiang could you please try to send the following response instead (all other fields are not necessary and may even disturb the processing - just to rule that possibility out):

{
        "topic":        "my.test/thing3/things/live/messages/TO--hello.world",
        "headers":      {
                "version":      2,
                "correlation-id":       "3711d8da-7677-491b-bffc-592f8a628b37",
                "content-type": "text/plain"
        },
        "value":        "I am a TeapotX",
        "status":       200,
        "path": "/outbox/messages/TO--hello.world"
}

thanks for your quick reply, i've tried , but it seems still get no response

[thjaeckle]

Ok, so to which address do you send the response via MQTT?
I guess to something like my.test/<something> because that is the only "source" address you configure.

For your questions:

1. is the ditto connection config is correct?
2. does Device correctly modify the request message ?
3. i'm not sure whether my client sdk call is correct, and which is right way?
4. how to set the timeout for client sdk
5. what topic should Device publish message to?

1. yes, seems ok
2. yes, looks also ok
3. yes, however I do miss one line which you should add: client.live().startConsumption().toCompletableFuture().get(); - I think without that you won't receive live messages with the client
4. use the timeout method:

   client.live().forId(ThingId.of("my.test", "thing3"))
                .message()
                .to()
                .subject("TO--hello.world")
                .timeout(Duration.ofSeconds(23)) // this one
                .payload("I am a TeapotX")
                .send(String.class, (response, throwable) -> {
                    System.out.println("XXXX: Got response: " + response.getPayload().orElse(null));  //<-----------Here, no any output is seen
                        }
                );

5. that is not defined by Ditto - you however have to add it to the sources of your MQTT connection

I hope that helps ..

[yuguanxiang]

it works now, thanks a lot !