Error with Kafka Topic

[Anonymous]

Hello,

I planed to use Talend Data Pipeline with Kafka.

I have few troubles with error in Dataset configuration. Connection with Kafka & topic is OK

With Kafka console, I can read data

Command

 docker-compose -f docker-compose-avro-connector.yaml exec connect /kafka/bin/kafka-console-consumer.sh   --bootstrap-server kafka:9092     --from-beginning     --formatter io.confluent.kafka.formatter.AvroMessageFormatter    --property schema.registry.url=http://schema-registry:8081     --topic sql.Sales.SpecialDeals

Datas

To see the whole post, download it here

[Anonymous]

Hello @pierre,

Thanks for trying Talend Pipeline Designer

If I understood it right, you have avro messages coming into your Kafka topic. You also retrieved the Avro schema through your Schema registry.

I would say that your schema looks correct as it comes directly from the schema registry. However, by seeing the indentation, did you copy-paste the whole result of your command ? 

This one: 

To see the whole post, download it here
OriginalPost.pdf

[Anonymous]

Hello Thibault,

I confirm that the shema specified in Talend Data Pipeline is only a part of the schema's Kafka.

I need to have some investigation because i'm not sure that shema registry was true.

For exemple, my datas has "before" node null by default and "after" node filled.

{
    "before": null,
    "after": {
        "csldebeizumsql.Sales.SpecialDeals.Value": {
            "SpecialDealID": 1,
            "StockItemID": null,
            "CustomerID": null,
            "BuyingGroupID": {
                "int": 2
            },
            "CustomerCategoryID": null,
            "StockGroupID": {
                "int": 7
            },
            "DealDescription": "Test Update",
            "StartDate": 16801,
            "EndDate": 16891,
            "DiscountAmount": null,
            "DiscountPercentage": {
                "bytes": "'\u0010"
            },
            "UnitPrice": null,
            "LastEditedBy": 2,
            "LastEditedWhen": 1451577600000000000
        }
    },
    "source": {
        "version": "1.0.0.Final",
        "connector": "sqlserver",
        "name": "csldebeizumsql",
        "ts_ms": 1579556959162,
        "snapshot": {
            "string": "true"
        },
        "db": "WideWorldImporters",
        "schema": "Sales",
        "table": "SpecialDeals",
        "change_lsn": null,
        "commit_lsn": {
            "string": "00000275:00002ff8:0001"
        },
        "event_serial_no": null
    },
    "op": "r",
    "ts_ms": {
        "long": 1579556959162
    }
}

I think shema was nos correct, I need to investigate 

[Anonymous]

Hello Thibaut,

I tried with Json formater and not avro

I delete this in my config connector

"key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter.schema.registry.url": "http://schema-registry:8081"

When I use console

docker-compose -f docker-compose-avro-connector.yaml exec connect /kafka/bin/kafka-console-consumer.sh   --bootstrap-server kafka:9092     --from-beginning        --topic csldebeizumsql.Sales.SpecialDeals

the result of data was this

To see the whole post, download it here
OriginalPost.pdf