Outbox Connector in standby state post deployment via terraform

[NehaShipt]

Hi team, I am deploying debezium (outbox and iceberg connectors ) on kafka connect service running on google kubernetes engine but whenever a new connector is deployed , the tasks are not getting initialized until the service webserver /worker pods are restarted , no error is shown in the logs.
below is my connect-distributed properties template:

Config Providers

config.providers=env
config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

==============================================================================

INFO {{.CommentLine}} is used to comment lines during development

group.id=kafka-connect-cluster-{{.AppEnv}}
bootstrap.servers={{.KafkaBrokers}}
reconnect.backoff.ms=3000
task.shutdown.graceful.timeout.ms=10000

10min, RECREATE

scheduled.rebalance.max.delay.ms=600000
listeners=HTTP://localhost:8083
rest.advertised.host.name={{.PodIP}}
rest.advertised.port={{.Port}}
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
config.storage.topic={{.ConfigTopic}}
offset.storage.topic={{.OffsetTopic}}
status.storage.topic={{.StatusTopic}}

plugins

plugin.path=/opt/kafka/plugins

==============================================================================

Workers Config

{{.CommentLine}}sasl.mechanism=PLAIN
{{.CommentLine}}security.protocol=SASL_SSL
{{.CommentLine}}sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
{{.CommentLine}} username="{{.KafkaKey}}"
{{.CommentLine}} password="{{.KafkaSecret}}";

==============================================================================

Connectors Config

connector.client.config.override.policy=All

Source connectors

{{.CommentLine}}producer.sasl.mechanism=PLAIN
{{.CommentLine}}producer.security.protocol=SASL_SSL
{{.CommentLine}}producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
{{.CommentLine}} username="{{.KafkaKey}}"
{{.CommentLine}} password="{{.KafkaSecret}}";

Sink connectors

{{.CommentLine}}consumer.sasl.mechanism=PLAIN
{{.CommentLine}}consumer.security.protocol=SASL_SSL
{{.CommentLine}}consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required
{{.CommentLine}} username="{{.KafkaKey}}"
{{.CommentLine}} password="{{.KafkaSecret}}";

Our resource allocation/ number of pods running/using strimzi kafka image are all good.
can anybody suggest what should I try to fix this issue?

Earlier we tried adding connect.protocol=eager in template but it makes all the connector in unassigned state though the tasks got initialized but again no use.

[scholzj]

A good start would be to:

• Make sure you share the full logs and custom resources (ideally including their status via kubect get ... -o yaml)
• Make sure you format what you share or attach it

Also, are you actually using Strimzi? The config you shared is badly formatted and pretty unreadable. But it does not look like Strimzi.

[NehaShipt]

Hi @scholzj Thanks for replying.

[NehaShipt]

yes we run kafka-connect as a service on google kuberneets engine and so uses strimzi behind the scene as the operator. The image I am using to build and deploy connect service is quay.io/strimzi/kafka:0.45.1-kafka-3.9.1
On top of it , I am adding debezium connector plugins to it debezium-connector-postgres:2.6.2](https://debezium.io/releases/2.6/#installation) , building it and deploying
The file I shared was connect-distributes.tmpl

# Config Providers
config.providers=env
config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

# ==============================================================================
# INFO {{.CommentLine}} is used to comment lines during development

group.id=kafka-connect-cluster-{{.AppEnv}}
bootstrap.servers={{.KafkaBrokers}}
reconnect.backoff.ms=3000
task.shutdown.graceful.timeout.ms=10000
# 10min, RECREATE
scheduled.rebalance.max.delay.ms=600000 

listeners=HTTP://localhost:8083
rest.advertised.host.name={{.PodIP}}
rest.advertised.port={{.Port}}

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

config.storage.topic={{.ConfigTopic}}
offset.storage.topic={{.OffsetTopic}}
status.storage.topic={{.StatusTopic}}

# plugins
plugin.path=/opt/kafka/plugins

# ==============================================================================
# Workers Config

{{.CommentLine}}sasl.mechanism=PLAIN
{{.CommentLine}}security.protocol=SASL_SSL
{{.CommentLine}}sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
{{.CommentLine}}  username="{{.KafkaKey}}" \
{{.CommentLine}}  password="{{.KafkaSecret}}";

# ==============================================================================
# Connectors Config

connector.client.config.override.policy=All

# Source connectors
{{.CommentLine}}producer.sasl.mechanism=PLAIN
{{.CommentLine}}producer.security.protocol=SASL_SSL
{{.CommentLine}}producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
{{.CommentLine}}  username="{{.KafkaKey}}" \
{{.CommentLine}}  password="{{.KafkaSecret}}";

# Sink connectors
{{.CommentLine}}consumer.sasl.mechanism=PLAIN
{{.CommentLine}}consumer.security.protocol=SASL_SSL
{{.CommentLine}}consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
{{.CommentLine}}  username="{{.KafkaKey}}" \
{{.CommentLine}}  password="{{.KafkaSecret}}"; 

[scholzj]

But in Strimzi, there is nothing like connect-distributes.tmpl. You configure it using the KafkaConnect and KafkaConnector resources, based on which the operator generates all the configs. So it is not really clear what your configuration refers to / what processes your template to generate the Strimzi resources. So you can for example try to do something like kubectl get strimzi -o yaml and share what you have there.

[NehaShipt]

Hi @scholzj Sorry for being late , I got little occupied with work. I will get this information today and share

[NehaShipt]

Hi @scholzj Apologies foe bring too late to reply, I got busy in some other thing but now I am back on this issue. I can share connect.yaml and Dockerfile for how we are building the connect image to deploy on GKE.

# [1/2] kafka-connect API
FROM golang:1.24.9 as builder
ENV CGO_ENABLED 0
ARG BUILD_REF=none

# copy cource, see .dockerignore
COPY . /app
WORKDIR /app/cmd/kafka-connect
RUN go build -ldflags "-X 'main.build=${BUILD_REF}'"

# [2/2] kafka-connect
FROM quay.io/strimzi/kafka:0.45.1-kafka-3.9.1

WORKDIR /opt/kafka

USER root
RUN chown -R kafka /opt/kafka/config
USER kafka

ENV JAVA_OPTS="-Xmx17G" \
    KAFKA_HEAP_OPTS="-Xmx17G" \
    KAFKA_JVM_PERFORMANCE_OPTS="-server -XX:MetaspaceSize=96m -XX:+UseG1GC -XX:MaxGCPauseMillis=50 -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80 -XX:+ExplicitGCInvokesConcurrent" \
    KAFKA_JMX_OPTS="-javaagent:./libs/jmx_prometheus_javaagent-1.0.1.jar=8081:config/jmx.yml -Djava.net.preferIPv4Stack=true"

COPY tmpl/ tmpl
COPY plugins/ plugins
COPY config/ config
COPY --from=builder /app/cmd/kafka-connect/kafka-connect .

CMD [ "./kafka-connect" 
```]


Would you like to see some special section of the connect.yaml as it contain some specific information to be shared on a public platform

[scholzj]

This is not how you use Strimzi. You seem to be just using the Strimzi container image as part of some pipeline, which I absolutely don't understand. But nothing you shared suggests you actually use the Strimzi operators, create the custom resources, etc.

[NehaShipt]

Yes we just use strimzi container image to build connect image for our service. I did not mean that we use strimzi directly , I took my question to kafka connect slack and those folks sent me here

[scholzj]

Yeah, but you are not really using Strimzi. You are just using the Kafka binaries that we downloaded from the Kafka website during the container image build. There seems to be nothing from Strimzi that you actually use. If you would use the Kafka Docker image, or build your own container image where you download the Kafka binaries, you would get exactly the same as you have right now. In fact, that is probably what I would recommend you to do, because unlike the Kafka container images, the Strimzi container images are designed with the single purpose of being used by the Strimzi operator. Or you should of course start using the Strimzi operators: https://strimzi.io/documentation/

The problem is that you are doing something complegtely custom but you are somehow expecting that we know what is it you are doing and can help you. But it is not the case. I have zero understanding of what you are actually doing beyond seeing that you do not use Strimzi.

Also, not sure where you asked, but I do not think Kafka has any Slack. It has only mailing lists: https://kafka.apache.org/contact

[NehaShipt]

So do you have some idea on this connect deployment running on kubernetes failing to rebalance tasks when a new connector is deployed until the service restarts

[NehaShipt]

@scholzj My bad if I was unable to explain the problem clearly. This issue is related to kafka-connect using debezium connector plugin. If you have some time and you think you can help me , I can send you a google meet or else could you point me to some places where I can ask for help.
https://debezium.zulipchat.com/#narrow/channel/348249-community-postgresql/topic/Debezium.20connectors.20not.20initializing.20task.20by.20itself/with/547936627 (because of this post I came to strimzi)

[NehaShipt]

@scholzj Just wanted to let you know that I found the issue , the correct value for listeners and port when running more than 1 replicas of connect in kubernetes are as below:
listeners=HTTP://0.0.0.0:8083
rest.advertised.host.name={{.PodIP}}
rest.advertised.port=8083

yes this is not a part of strimzi but can be helpful for anyone else who have a similar use-case.