Adding custom transformer plugin doesn't work.

[fathimaSheikh]

I'm trying to run a kafka connect cluster for confluent S3SinkConnector using Strimzi operator. I have some custom Transformations which I'm trying to add on top of Strimzi docker image, but the operator complains that no such transform class was found.
In the connector deployment logs, I see my Transformation class being loaded but the Strimzi operator logs complain that no such class is found.

Here's my docker file:

FROM --platform=linux/amd64 confluentinc/cp-kafka-connect:7.3.2 as cp
ARG GITHUB_PACKAGES_TOKEN
RUN confluent-hub install --no-prompt confluentinc/kafka-connect-s3:10.4.2
RUN confluent-hub install --no-prompt confluentinc/connect-transforms:1.4.3
RUN wget -P /tmp https://myorg-github-robot:[email protected]/mine/kafka-connect-transformer/me/mine/kafka-connect-transformer/0.10.0/kafka-connect-transformer-0.10.0.jar


FROM quay.io/strimzi/kafka:latest-kafka-3.0.0

USER root:root


RUN mkdir -p /opt/kafka/plugins/kafka-connect-s3 && \
 mkdir -p /opt/kafka/plugins/connect-transforms



COPY --from=cp /usr/share/confluent-hub-components/confluentinc-kafka-connect-s3/lib /opt/kafka/plugins/kafka-connect-s3/
COPY --from=cp /usr/share/confluent-hub-components/confluentinc-connect-transforms/lib /opt/kafka/plugins/connect-transforms/
COPY --from=cp /tmp/kafka-connect-transformer-0.10.0.jar /opt/kafka/plugins/connect-transforms/kafka-connect-transformer-0.10.0.jar

USER 1001

Here are logs from connector deployed.

2023-05-09 13:59:02,098 INFO Scanning for plugin classes. This might take a moment ... (org.apache.kafka.connect.cli.ConnectDistributed) [main]
2023-05-09 13:59:02,114 INFO Loading plugin from: /opt/kafka/plugins/connect-transforms (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Registered loader: PluginClassLoader{pluginLocation=file:/opt/kafka/plugins/connect-transforms/} (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'me.mine.kafka.connect.transforms.TombstoneTransformer$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.TombstoneHandler' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.ExtractTopic$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.Filter$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.ExtractTopic$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.ExtractTopic$Header' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,348 INFO Added plugin 'io.confluent.connect.transforms.GzipDecompress$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'io.confluent.connect.transforms.Drop$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'me.mine.kafka.connect.transforms.TombstoneTransformer$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'io.confluent.connect.transforms.Filter$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'io.confluent.connect.transforms.MessageTimestampRouter' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'io.confluent.connect.transforms.GzipDecompress$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO Added plugin 'io.confluent.connect.transforms.Drop$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,349 INFO **Added plugin 'me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]**
2023-05-09 13:59:02,349 INFO **Added plugin 'me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Key' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]**
2023-05-09 13:59:02,350 INFO Added plugin 'org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,350 INFO Added plugin 'org.apache.kafka.connect.connector.policy.PrincipalConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]
2023-05-09 13:59:02,350 INFO Added plugin 'org.apache.kafka.connect.connector.policy.NoneConnectorClientConfigOverridePolicy' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader) [main]

Here's my Strimzi_operator yaml:

# Default values for strimzi-kafka-operator.

# Default replicas for the cluster operator
replicas: ${strimzi_kafka_operator_replicas}

# If you set `watchNamespaces` to the same value as ``.Release.Namespace` (e.g. `helm ... --namespace $NAMESPACE`),
# the chart will fail because duplicate RoleBindings will be attempted to be created in the same namespace
watchNamespaces: []
watchAnyNamespace: true

defaultImageRegistry: quay.io
defaultImageRepository: strimzi
defaultImageTag: ${strimzi_kafka_operator_image_tag}

image:
  registry: ""
  repository: ""
  name: operator
  tag: ""
logVolume: co-config-volume
logConfigMap: strimzi-cluster-operator
logConfiguration: ""
fullReconciliationIntervalMs: 120000
operationTimeoutMs: 300000
kubernetesServiceDnsDomain: cluster.local
featureGates: ""
tmpDirSizeLimit: 1Mi


# Example on how to configure extraEnvs
# extraEnvs:
#   - name: JAVA_OPTS
#     value: "-Xms256m -Xmx256m"

extraEnvs: []

tolerations: []
affinity: {}
annotations: {}
labels: {}
nodeSelector: {}
priorityClassName: ""

podSecurityContext: {}
securityContext:
  allowPrivilegeEscalation: false
  seccompProfile:
    type: RuntimeDefault
  runAsNonRoot: true
  capabilities:
    drop:
      - ALL
rbac:
  create: yes
serviceAccountCreate: yes
serviceAccount: strimzi-cluster-operator

kafkaConnect:
  image:
    registry: ${ecr_registry}
    repository: ${ecr_repository_kafka_connect}
    name: ${ecr_image_name_kafka_connect}
    tag: ${ecr_image_tag_kafka_connect}

resources:
  limits:
    memory: 384Mi
    cpu: 1000m
  requests:
    memory: 384Mi
    cpu: 200m
livenessProbe:
  initialDelaySeconds: 10
  periodSeconds: 30
readinessProbe:
  initialDelaySeconds: 10
  periodSeconds: 30

createGlobalResources: true
# Create clusterroles that extend existing clusterroles to interact with strimzi crds
# Ref: https://kubernetes.io/docs/reference/access-authn-authz/rbac/#aggregated-clusterroles
createAggregateRoles: false
# Override the exclude pattern for exclude some labels
labelsExclusionPattern: ""
# Controls whether Strimzi generates network policy resources (By default true)
generateNetworkPolicy: true
# Override the value for Connect build timeout
connectBuildTimeoutMs: 300000

And finally the connector.yaml

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: "${connector_name}-s3-sink-connector"
  namespace: ${namespace}
  labels:
    strimzi.io/cluster: "${connect_cluster_name}"
spec:
  autoRestart:
    enabled: true
  class: io.confluent.connect.s3.S3SinkConnector
  tasksMax: ${num_tasks}
  config:
     "consumer.auto.offset.reset": "none"
     "flush.size": "5000"
     "format.class": "io.confluent.connect.s3.format.json.JsonFormat"
     "key.converter": "org.apache.kafka.connect.storage.StringConverter"
     "locale": "de_DE"
     "name": "${connector_name}-s3-sink-connector"
     "partition.duration.ms": "3600000"
     "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner"
     "path.format": "'partition_date'=YYYY-MM-dd/'hour'=HH"
     "rotate.interval.ms": "${rotate_interval_ms}"
     "s3.bucket.name": "${bucket_name}"
     "s3.compression.type": "gzip"
     "s3.region": "eu-central-1"
     "s3.compression.level": 1
     "s3.part.size": "5242880"
     "schemas.enable": "false"
     "storage.class": "io.confluent.connect.s3.storage.S3Storage"
     "timezone": "UTC"
     "topics": "events_v2"
     "topics.dir": "public/raw"
     "timestamp.extractor" : "me.mine.kafka.connect.storage.partitioner.HeaderFieldTimestampExtractor"
     "value.converter": "org.apache.kafka.connect.json.JsonConverter"
     "value.converter.schemas.enable": "false"
     "transforms" : "offerListExtractHeaderTransformer"
     "transforms.offerListExtractHeaderTransformer.header.key.eventCreationTime" : "offerListEventTime"
     "transforms.offerListExtractHeaderTransformer.type" : "me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value"
     "errors.tolerance": "none"
     "errors.logs.enable": "true"
     "errors.log.include.messages": "true"

Error logs from Strimzi operator:

Reconciliation #1(watch) KafkaConnect(ingestion/ingestion-cluster): Error reconciling connector ingestion-events-v2-s3-sink-connector
io.strimzi.operator.cluster.operator.assembly.ConnectRestException: PUT /connectors/ingestion-events-v2-s3-sink-connector/config returned 400 (Bad Request): Connector configuration is invalid and contains the following 2 error(s):
Invalid value me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value for configuration transforms.offerListExtractHeaderTransformer.type: Class me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value could not be found.
Invalid value null for configuration transforms.offerListExtractHeaderTransformer.type: Not a Transformation
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`
	at io.strimzi.operator.cluster.operator.assembly.KafkaConnectApiImpl.lambda$createOrUpdatePutRequest$1(KafkaConnectApiImpl.java:92) ~[io.strimzi.cluster-operator-0.34.0.jar:0.34.0]

Could someone help me understand what's wrong with my configuration for Strimzi?
maybe point me to some documentation about adding custom transformations plugin.

[fathimaSheikh]

Looks like the image was not picked up right. After starting from scratch, the above issue is resolved now. And now I see a different problem.

Error reconciling connector ingestion-events-v2-s3-sink-connector
io.strimzi.operator.cluster.operator.assembly.ConnectRestException: PUT /connectors/ingestion-events-v2-s3-sink-connector/config returned 400 (Bad Request): Connector configuration is invalid and contains the following 1 error(s):
Invalid value class me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value for configuration transforms.offerListExtractHeaderTransformer.type: Error getting config definition from Transformation: Could not instantiate class me.mine.kafka.connect.transforms.ExtractHeaderTransformer$Value
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`

Any ideas what's wrong with the connector.yaml? Tried adding an escape character"
"transforms.offerListExtractHeaderTransformer.type" : "me.mine.kafka.connect.transforms.ExtractHeaderTransformer\$Value"

But I get error found unknown escape character