suisenkotoba
commented
7 months ago topic config
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
annotations:
kubectl.kubernetes.io/last-applied-configuration: |
{"apiVersion":"kafka.strimzi.io/v1beta2","kind":"KafkaTopic","metadata":{"annotations":{},"labels":{"strimzi.io/cluster":"kafka01"},"name":"loan-logs---184077157bfc68094e73c105123319efa806b1f3","namespace":"kafka-prod"},"spec":{"config":{"cleanup.policy":"delete","max.message.bytes":"1048588","min.insync.replicas":"1","retention.bytes":"-1","retention.ms":"1209600000"},"partitions":10,"replicas":3,"topicName":"loan_logs"}}
creationTimestamp: "2023-11-09T03:34:54Z"
generation: 5
labels:
strimzi.io/cluster: kafka01
name: loan-logs---184077157bfc68094e73c105123319efa806b1f3
namespace: kafka-prod
resourceVersion: "141694147"
uid: 1936b625-2603-402c-be22-1d7333cfd029
spec:
config:
cleanup.policy: delete
max.message.bytes: "1048588"
message.format.version: 3.0-IV1
min.insync.replicas: "1"
retention.bytes: "-1"
retention.ms: "1209600000"
partitions: 10
replicas: 3
topicName: loan_logs
status:
conditions:
- lastTransitionTime: "2023-11-22T09:28:54.111074277Z"
status: "True"
type: Ready
observedGeneration: 5
topicName: loan_logssink connector:
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
labels:
strimzi.io/cluster: gcs-connect-cluster
name: de-lsm-gcs-connector
namespace: kafka-operator
spec:
class: io.aiven.kafka.connect.gcs.GcsSinkConnector
config:
topics: loan_logs
file.name.prefix: lsm/
gcs.credentials.path: /opt/kafka/external-configuration/gcs-auth-secret/google_credentials.json
gcs.bucket.name: datalake-raw
file.name.timestamp.timezone: Asia/Jakarta
file.name.template: "{{topic}}/sink_date={{timestamp:unit=yyyy}}{{timestamp:unit=MM}}{{timestamp:unit=dd}}/sink_hour={{timestamp:unit=HH}}/p{{partition:padding=false}}-{{start_offset:padding=true}}.snappy.parquet"
file.compression.type: snappy
format.output.fields: key,value,offset,timestamp,headers
format.output.type: parquet
format.output.envelope: true
behavior.on.null.values: ignore
key.converter: org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable: false
value.converter: org.apache.kafka.connect.storage.StringConverter
value.converter.schemas.enable: false
tasksMax: 4I forgot to mention earlier, we set up a cron job to pause-resume kafka connector for every 15 minutes to avoid small file problem in GCS. This is for cron jobs:
---
apiVersion: batch/v1
kind: CronJob
metadata:
name: de-lsm-gcs-connector-pause
namespace: kafka-prod
spec:
schedule: "15,30,45,00 * * * *"
jobTemplate:
spec:
template:
spec:
containers:
- name: de-lsm-gcs-connector-pause
image: quay.io/curl/curl:8.6.0
imagePullPolicy: IfNotPresent
command:
- /usr/bin/curl
- -s
- -X
- PUT
- http://gcs-connect-cluster-connect-api:8083/connectors/de-lsm-gcs-connector/pause
restartPolicy: OnFailure
---
apiVersion: batch/v1
kind: CronJob
metadata:
name: de-lsm-gcs-connector-resume
namespace: kafka-prod
spec:
schedule: "14,29,44,57 * * * *"
jobTemplate:
spec:
template:
spec:
containers:
- name: de-lsm-gcs-connector-resume
image: quay.io/curl/curl:8.6.0
imagePullPolicy: IfNotPresent
command:
- /usr/bin/curl
- -s
- -X
- PUT
- http://gcs-connect-cluster-connect-api:8083/connectors/de-lsm-gcs-connector/resume
restartPolicy: OnFailureEven though there is job to pause the connector, it did not seem to affect the overwriting. The connector would be in state PAUSED briefly then start running again before resume job started and the overwriting continue.
I did say there were overwritten files with 0 byte size, but that was the case in dev environment. In production the files are not 0 bytes, but they are still got overwritten every minute. Since the filename has partition number prefix and ended with offset, I notice the offset got repeated in the same partition but different hour. and when I check the content there were indeed duplicate key with same offset. The files in GCS bucket are organized by sink date and sink hour, so different hour different object prefix. I think if there is not any prefix by sink date/hour, there would not be any duplicate.
(files in sink hour 9)
(files in sink hour 10)
jeqo
Some of the files written by kafka sink connector in bucket has 0 bytes, I wonder if this is normal behaviour?
In another hour all of the files have 0 bytes and I notice they keep getting overwritten every minute until next hour and it keeps happening again.
I also found this error in connector logs, not sure if they are related with the issue:
aiven version:
0.13.0kafka version:3.5.1this is my connector configuration:I am using the same configuration in different kafka with version
3.3and aiven0.9.0and there has not been any issue similar to this. Is there any new configuration I am missing?