'How to get org.apache.kafka.connect.data.Decimal value from Kafka JSON message [duplicate]

I use debizium to stream postgresql data to Kafka, and use Java to subscribe Kafka topic.

I receive Kafka message and get a JSON string, but one numeric value can not recognized.

The JSON is:

{
    "schema":
    {
        "type": "struct",
        "fields": [
        {
            "type": "struct",
            "fields": [
            {
                "type": "string",
                "optional": true,
                "field": "creator"
            },
            {
                "type": "int64",
                "optional": true,
                "name": "io.debezium.time.MicroTimestamp",
                "version": 1,
                "field": "createtime"
            },
            {
                "type": "bytes",
                "optional": true,
                "name": "org.apache.kafka.connect.data.Decimal",
                "version": 1,
                "parameters":
                {
                    "scale": "5",
                    "connect.decimal.precision": "32"
                },
                "field": "familyprice"
            }],
            "optional": true,
            "name": "pssdev.public.order.Value",
            "field": "before"
        },
        {
            "type": "struct",
            "fields": [
            {
                "type": "string",
                "optional": true,
                "field": "creator"
            },
            {
                "type": "int64",
                "optional": true,
                "name": "io.debezium.time.MicroTimestamp",
                "version": 1,
                "field": "createtime"
            },
            {
                "type": "bytes",
                "optional": true,
                "name": "org.apache.kafka.connect.data.Decimal",
                "version": 1,
                "parameters":
                {
                    "scale": "5",
                    "connect.decimal.precision": "32"
                },
                "field": "familyprice"
            }],
            "optional": true,
            "name": "pssdev.public.order.Value",
            "field": "after"
        },
        {
            "type": "struct",
            "fields": [
            {
                "type": "string",
                "optional": true,
                "field": "version"
            },
            {
                "type": "string",
                "optional": true,
                "field": "connector"
            },
            {
                "type": "string",
                "optional": false,
                "field": "name"
            },
            {
                "type": "string",
                "optional": false,
                "field": "db"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "ts_usec"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "txId"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "lsn"
            },
            {
                "type": "string",
                "optional": true,
                "field": "schema"
            },
            {
                "type": "string",
                "optional": true,
                "field": "table"
            },
            {
                "type": "boolean",
                "optional": true,
                "default": false,
                "field": "snapshot"
            },
            {
                "type": "boolean",
                "optional": true,
                "field": "last_snapshot_record"
            },
            {
                "type": "int64",
                "optional": true,
                "field": "xmin"
            }],
            "optional": false,
            "name": "io.debezium.connector.postgresql.Source",
            "field": "source"
        },
        {
            "type": "string",
            "optional": false,
            "field": "op"
        },
        {
            "type": "int64",
            "optional": true,
            "field": "ts_ms"
        }],
        "optional": false,
        "name": "pssdev.public.order.Envelope"
    },
    "payload":
    {
        "before":
        {
            "creator": null,
            "createtime": null,
            "familyprice": null
        },
        "after":
        {
            "creator": "USER1E",
            "createtime": 1554292597815000,
            "familyprice": "W42A"
        },
        "source":
        {
            "version": "0.9.5.Final",
            "connector": "postgresql",
            "name": "pssdev",
            "db": "pf",
            "ts_usec": 1561459811737920,
            "txId": 771604,
            "lsn": 88282458880,
            "schema": "public",
            "table": "order",
            "snapshot": false,
            "last_snapshot_record": null,
            "xmin": null
        },
        "op": "u",
        "ts_ms": 1561459811747
    }
}

The familyprice value is W42A I can't figure out how to convert it.

The actual familyprice value in database is 60.00000, the column type is numeric(32,5)

javaapache-kafkadecimaldebezium


Solution 1:[1]

Please see https://debezium.io/documentation/faq/#how_to_retrieve_decimal_field_from_binary_representation for Java clients.

You can also set decimal.handling.mode to different value so you'll reciev the Decimal either as string or double if it easier for you.

Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source
Solution 1 OneCricketeer

Related Questions