Integration of the data plane Kafka extension

[boxer-code]

Hello everyone!

I'm currently trying to use the data plane kafka extension. Has anyone an example how to configure and especially how to use it? Do I simply need to start a connector with this extension and with the parameters for the topic and kafka server configured in the configuration file? How can I start the communication between the connector and the kafka kraft server? Is it similar to the sample 06?

Thanks in advance!

[boxer-code]

I've tried the following body for a transfer request and allowed "Kafka" as a type at the steps before. But I'm struggling with the following error:

DEBUG 2023-07-19T14:24:23.926038 TransferProcess: ID 792f166b-fe99-4101-8a99-32c9ec54d48d. Initiate data flow
SEVERE 2023-07-19T14:24:23.926466 TransferProcess: ID 792f166b-fe99-4101-8a99-32c9ec54d48d. Fatal error while Initiate data flow. Error details: Failed to delegate data transfer to Data Plane: Missing or invalid value for key topic, Missing `bootstrap.servers` config
WARNING 2023-07-19T14:24:23.926559 Failed to delegate data transfer to Data Plane: Missing or invalid value for key topic, Missing `bootstrap.servers` config
DEBUG 2023-07-19T14:24:23.926699 TransferProcess 792f166b-fe99-4101-8a99-32c9ec54d48d is now in state TERMINATING

The request looks like this:

{
        "@context": {
          "edc": "https://w3id.org/edc/v0.0.1/ns/"
        },
        "@type": "TransferRequestDto",
        "connectorId": "provider",
        "connectorAddress": "http://localhost:19194/protocol",
        "contractId": "{{contract agreement}}",
        "assetId": "assetId",
        "managedResources": false,
        "protocol": "dataspace-protocol-http",
        "dataDestination": { 
          "type": "Kafka",
          "baseUrl": "http://localhost:9092"
        }
    }

And my configuration file looks like this:

edc.participant.id=consumer
edc.ids.id=urn:connector:consumer
edc.dsp.callback.address=http://localhost:29194/protocol
web.http.port=29191
web.http.path=/api
web.http.management.port=29193
web.http.management.path=/management
web.http.protocol.port=29194
web.http.protocol.path=/protocol
edc.receiver.http.endpoint=http://localhost:4000/receiver/urn:connector:provider/callback
edc.public.key.alias=public-key
edc.transfer.dataplane.token.signer.privatekey.alias=1
edc.transfer.proxy.token.signer.privatekey.alias=1
edc.transfer.proxy.token.verifier.publickey.alias=public-key
web.http.public.port=29291
web.http.public.path=/public
web.http.control.port=29192
web.http.control.path=/control
edc.dataplane.token.validation.endpoint=http://localhost:29192/control/token
edc.dataplane.kafka.topic="topic"
edc.kafka.bootstrap.servers="http://localhost:9092"
edc.dataplane.kafka.bootstrap.servers="http://localhost:9092"
edc.bootstrap.servers="http://localhost:9092"

What am I missing?

[ghost]

Eventhough this answer comes a bit late, future visitors should consider reviewing the following sample when in search for answers related with kafka extension configuration.
https://github.com/eclipse-edc/Samples/tree/main/transfer/streaming/streaming-02-kafka-to-http