{{ source:address }} connection placeholder

[neottil]

For the documentation is possible to configure enforcement input using placeholder {{ source:address }}. https://www.eclipse.org/ditto/3.2/basic-connections.html#source-enforcement

I try to configure Kafka connection like this:
{
"id": "glin_dl00001_prod-kafka-connection",
"name": null,
"connectionType": "kafka",
"connectionStatus": "open",
"uri": "tcp://kafka-service:9092",
"sources": [
{
"enforcement": {
"input": "{{ source:address }}",
"filters": "{{ thing:namespace }}.kafakinbound"
},
"addresses": [
"glin_dl00001_dev.kafkainbound",
"glin_dl00001_prod.kafkainbound"
],
"consumerCount": 1,
"qos": 0,
"authorizationContext": [
"device:glin_dl00001_prod"
],
"headerMapping": {
"content-type": "{{header:content-type}}",
"correlation-id": "{{header:correlation-id}}"
},
"replyTarget": {
"address": "{{thing:namespace}}.command.inbound/{{thing:namespace}}/{{thing:name}}",
"headerMapping": {
"content-type": "{{header:content-type}}",
"correlation-id": "{{header:correlation-id}}"
},
"expectedResponseTypes": [
"response",
"error"
],
"enabled": true
}
}
],
"targets": [
{
"address": "{{thing:namespace}}.command.inbound/{{thing:namespace}}/{{thing:name}}",
"topics": [
"//things/twin/events",
"//things/live/messages"
],
"qos": 0,
"authorizationContext": [
"nginx:user-pipeline"
],
"headerMapping": {}
}
],
"clientCount": 1,
"failoverEnabled": true,
"validateCertificates": true,
"processorPoolSize": 1,
"specificConfig": {
"bootstrapServers": "kafka-service:9092"
},
"tags": []
}

The response of connection API is:
{
"status": 400,
"error": "connectivity:connection.configuration.invalid",
"message": "The placeholder 'source:address' could not be resolved.",
"description": "Some placeholders could not be resolved. Check the spelling of the placeholder and make sure all required headers are set."
}

This is not supported for Kafka connections?

[thjaeckle]

Hi @neottil

I just checked .. Unfortunately, the source:address placeholder currently only is supported for MQTT connections:

ditto/connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/placeholders/ConnectivityPlaceholders.java

Line 96 in 6f7a336

public static SourceAddressPlaceholder newSourceAddressPlaceholder() {

This is currently only used in:

ditto/connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/AbstractMqttValidator.java

Line 129 in 6f7a336

final SourceAddressPlaceholder sourceAddressPlaceholder = newSourceAddressPlaceholder();

I however don't see why this should only be available for MQTT, would also make sense for other connections.

[neottil]

Seems in KafkaValidator.java are checked headers only.

ditto/connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/kafka/KafkaValidator.java

Line 108 in 6f7a336

validateTemplate(enforcement.getInput(), dittoHeaders, PlaceholderFactory.newHeadersPlaceholder());

Do you think we can have the fix for this bug in a next version?

[neottil]

Is it possible to open an issue to track the bug?

[thjaeckle]

Sure, please go ahead ..