Open andyhuynh3 opened 7 months ago
the feature is available from confluent-schema-registry release 7.3 are you using this version of the ser/deser lib in your kafka-connect ?
check the doc -> https://docs.confluent.io/platform/current/schema-registry/connect.html#null-values-replaced-with-default-values
@raphaelauv Yes I'm using version 7.6.0. It works on the producer side (e.g. with Debezium) but the issue is with the Confluent Kafka Connect sinks. I have a PR here to address the issue.
"value.converter.ignore.default.for.nullables": "true"
work with
FROM confluentinc/cp-kafka-connect-base:7.6.0
RUN (echo 1 && yes) |confluent-hub install confluentinc/kafka-connect-s3:10.5.0
and the final file drop on s3 contain
{"currency":null,"contry":"UNKNOW","_kafka_partition":0,"_kafka_offset":9,"_kafka_timestamp":1707851123401}
I'm not using the Confluent Kafka Connect image to begin with, but I can give it a try when I have some time.
More details on my setup -- I'm working with the Strimzi base image (quay.io/strimzi/kafka:0.38.0-kafka-3.5.1
) and installing version 10.5.7 of the S3 sink connector:
ARG KAFKA_VERSION="3.5.1"
ARG STRIMZI_VERSION="0.38.0"
FROM quay.io/strimzi/kafka:${STRIMZI_VERSION}-kafka-${KAFKA_VERSION}
ARG CONFLUENT_S3_SINK_VERSION="10.5.7"
ENV CONFLUENT_S3_SINK_VERSION=${CONFLUENT_S3_SINK_VERSION}
RUN mkdir -p /opt/kafka/plugins/s3
RUN curl https://d1i4a15mxbxib1.cloudfront.net/api/plugins/confluentinc/kafka-connect-s3/versions/${CONFLUENT_S3_SINK_VERSION}/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}.zip -o /tmp/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}.zip
RUN cd /tmp && unzip /tmp/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}.zip
RUN cp -r /tmp/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}/lib /opt/kafka/plugins/s3/
RUN rm /tmp/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}.zip && rm -rf /tmp/confluentinc-kafka-connect-s3-${CONFLUENT_S3_SINK_VERSION}
I do have "value.converter.ignore.default.for.nullables": "true"
with this setup, but I still get nulls being replaced with the default in the S3 files.
Check the version of the schema-registry libs jar in that container image
Hello, I'm using Debezium to extract MySQL data into Kafka in Avro format using the Confluent Avro converter. I'm then using the Confluent S3 sink to get this data into S3 as Avro files. However I'm running into an issue on the Kafka --> S3 side where my null values are being replaced with the MySQL default, even with
value.converter.ignore.default.for.nullables=true
. More details on setup below:Here's what my S3 sink settings look like
Here's what my schema looks like:
And here's what my message looks like in Kafka:
And when I try to read the Avro file produced by the S3 connector via Python, this is what I'm seeing
Notice how the value for the
test_str
key is the default value (alsotest_str
) instead ofNone
or null.In part of the S3 connector logs, I do see
ignore.default.for.nullables = false
, so is this setting perhaps not taking?