debezium-connector for ibmi (AS/400)

[aqhil07]

@msillence @martintosney

Hello All,

We are trying to implement Debezium connector for IBMi. Found below link and would like to understand below points.

https://github.com/jhc-systems/debezium-connector-ibmi

1. Does this connector suitable for Production environment.
2. Will any support provided for connector.
3. Docker image available in docker hub ?

Thanks,
Akhil.

[msillence]

Hi Akhil

1/ we are using the connector in production
2/ I don't offer any formal support but there are moves to bring this under the debezium main project as others are also interested in using this. In the past I have fixed reported issues but that likely depends on the issue and it's alignment with our use and our ability to reproduce/support it.
3/ on the releases tab I tend to include the docker images https://github.com/jhc-systems/debezium-connector-ibmi/releases

Hope you find a suitable solution.

[aqhil07]

Thanks Martin for the update, we will try and update here. If any problem occurs.

[aqhil07]

Hi Martin,

Can we follow Debezium connector for Db2 steps and connector configuration files for db2 i as well. I'll use docker image file from here (docker.io/silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.4). Will that work ?

https://debezium.io/documentation/reference/stable/connectors/db2.html

[msillence]

I'd suggest looking at the docker compose and then use the instructions under configure-the-cdc section of the readme
https://github.com/jhc-systems/debezium-connector-ibmi#configuring-the-cdc

I'm not sure much of the DB2 connector is relevant

There are also some instructions on the permissions for the account: https://github.com/jhc-systems/debezium-connector-ibmi/tree/main/journal-parsing#permissions

I should really restructure this to make it easier to get up and running

[aqhil07]

Sure Thanks Martin will have a look on above links. On the other side, is there any idea when ibm i would involve into Debezium project.

[ash1912sha]

hi martin,

in the below configuration , I did not find the Kafka topics details . Please let me know

{
"name": "connector-name",
"config": {
"connector.class": "io.debezium.connector.db2as400.As400RpcConnector",
"schema": "SCHEMA",
"sanitize.field.names": "true",
"tasks.max": "1",
"hostname": "ibmiserver",
"dbname": "database name",
"user": "xxx",
"password": "xxx",
"port": "",
"secure": true
"poll.interval.ms": "2000",
"transforms": "unwrap",
"transforms.unwrap.delete.handling.mode": "rewrite",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"table.include.list": "SCHEMA.TABLE1",
"snapshot.mode": "initial",

"key.converter.schema.registry.url": "http://schema-registry:8081",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",

"snapshot.max.threads": 4

}
}

[msillence]

by the "topics" do you mean the tables you want to stream - they are in the "table.include.list": "SCHEMA.TABLE1",

[msillence]

If you meant the kafka bootstrap server then that goes in the environment variables see the docker compose file

debezium-connector-ibmi/docker-compose.yaml

Line 73 in 2698ed8

- DEBEZIUM_BOOTSTRAP_SERVERS=mskafka:9092

[ash1912sha]

no i mean.

if i am streaming data from db2 tables then in configuration file i don't see topic variable where i can give topic name. but i see db2 details like below.
"hostname": "ibmiserver",
"dbname": "database name",
"user": "xxx",
"password": "xxx",
"port": "",
"table.include.list": "SCHEMA.TABLE1"

[ash1912sha]

like in which topic i should insert data in kafka

[msillence]

it is automatic, the topic name will be <hostname>.<schema>.<table> you can use a SMT to write the table name
I think there is a slight deviation from the standard debezium output of <dbname>.<schema>.<table> which might change when debezium integrate this

[ash1912sha]

Hi,

actually our Kafka team does not allow to create topic automatically. So if we create topics manually as per mentioned (..

) so the question is our hostname itself has "." in between name. how connector will distinguish the hostname and db and table

our db2 sample details

hostname - SAMPLE.SERVER.TEST.NET
database - PDATABASEL
table - TABLEQO

Please let us know what shall we give the topic name in Kafka to resolve this problem

[msillence]

topic name would be SAMPLE.SERVER.TEST.NET.PDATABASEL.TABLEQO

if the debezium connector runs without creat permissions I guess you'd see an error in the logs for the topic it wanted

[ash1912sha]

if we leave the below property as true and create topic as mentioned by you SAMPLE.SERVER.TEST.NET.PDATABASEL.TABLEQO manually then will it give the error ?
auto.create.topics.enable=true

[ash1912sha]

Hi martin,

if you can update on this ?

[aqhil07]

@msillence Just Wondering do we also require Schema Change topic and Schema history topic as well. By Default all Debezium projects those are having CDC capability requires these topics. Can you please clarify ?

[msillence]

I think how I've implemented this these two are missing, it's something that came up in the discussion with the Jiri and we should be able to support it but I'm not sure when it will happen.

[KanchanJoyia]

@msillence we are trying to deploy the Debezium ibmi connnector. However we are not able to copy the image from below path
I am using docker image file from here (docker.io/silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.4). its not working .

[msillence]

the latest is actually docker.io/silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.5
The list of images can be seen here: https://hub.docker.com/r/silllencem/debezium-connector-ibmi/tags
I can pull the image you mention

[ash1912sha]

i am getting error while running the below commands for docker image.

FROM silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.5
ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"
ENV JAVA_TOOL_OPTIONS="-Dhttps.proxyHost=app-proxy.localendpoint.banksvcs.net -Dhttps.proxyPort=8080 -Dhttp.proxyHost=app-proxy.localendpoint.banksvcs.net -Dhttp.proxyPort=8080 -Dhttp.nonProxyHosts=localhost|eu-west-1.amazonaws.com"
RUN silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.5
ENV JAVA_TOOL_OPTIONS="-Dhttps.proxyHost= -Dhttps.proxyPort= -Dhttp.proxyHost= -Dhttp.proxyPort= -Dhttp.nonProxyHosts="

[msillence]

Sorry but I'm not sure I see any errors just commands from a Dockerfile?

Notably I don't think the run command looks right I don't think you can RUN a docker image from within a Dockerfile

I'd guess from the connector_plugin_path you are thinking about copying files from one docker image to another?
Only two images are in the above and they are both the same one - the FROM and the RUN lines

If you wanted to add this debezium connector to a confluent image say then you probably need to do something more like this:
https://www.linkedin.com/pulse/how-copy-docker-image-from-one-machine-another-abhishek-rana/

As mentioned in this issue #70 I'm afraid I don't have time to support this endevour

If you want to simply run the published images then you probably need a command like

docker run --env-file environment-file.txt silllencem/debezium-connector-ibmi:2.4.0.Final_1.10.5

where you need to create the environment file with the parameters listed in the readme or take a look at the docker-compose.yaml which can also spin up a kafka instance for you https://github.com/jhc-systems/debezium-connector-ibmi/blob/main/docker-compose.yaml

[ash1912sha]

Hi Martin,

I am using below setting and my pod is not coming up. Could you please help me here.

#commented one more time
apiVersion: apps/v1
kind: Deployment
metadata:
name: db2i-connector
labels:
{{- include "db2i-connect-cluster.labels" . | nindent 4 }}
spec:
{{- if not .Values.autoscaling.enabled }}
replicas: {{ .Values.replicaCount }}
{{- end }}
selector:
matchLabels:
{{- include "db2i-connect-cluster.selectorLabels" . | nindent 6 }}
template:
metadata:
{{- with .Values.podAnnotations }}
annotations:
{{- toYaml . | nindent 8 }}
{{- end }}
labels:
{{- include "db2i-connect-cluster.selectorLabels" . | nindent 8 }}
spec:
{{- with .Values.imagePullSecrets }}
imagePullSecrets:
{{- toYaml . | nindent 8 }}
{{- end }}
serviceAccount: db2i-connect-cluster
volumes:
- name: kafka-server-cert
csi:
driver: secrets-store.csi.k8s.io
readOnly: true
volumeAttributes:
secretProviderClass: "kafka-server-cert"
securityContext:
{{- toYaml .Values.podSecurityContext | nindent 8 }}
containers:
- name: db2i-connector
volumeMounts:
- name: kafka-server-cert
mountPath: "/mnt/sslcerts/"
readOnly: false
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
env:
- name: DEBEZIUM_CONFIG_STORAGE_TOPIC
value: {{ .Values.config_storage_topic }}
- name: DEBEZIUM_STATUS_STORAGE_TOPIC
value: {{ .Values.status_storage_topic }}
- name: DEBEZIUM_OFFSET_STORAGE_TOPIC
value: {{ .Values.offset_storage_topic }}
- name: DEBEZIUM_CONSUMER_GROUP_ID
value: {{ .Values.consumer_group_id }}
- name: DEBEZIUM_GROUP_ID
value: {{ .Values.group_id }}
- name: DEBEZIUM_BOOTSTRAP_SERVERS
value: {{ .Values.broker_url }}
- name: DEBEZIUM_KEY_CONVERTER_SCHEMAS_ENABLE
value: "false"
- name: DEBEZIUM_VALUE_CONVERTER_SCHEMAS_ENABLE
value: "false"
- name: DEBEZIUM_KEY_CONVERTER
value: "io.confluent.connect.avro.AvroConverter"
- name: DEBEZIUM_VALUE_CONVERTER
value: "io.confluent.connect.avro.AvroConverter"
- name: PARTITIONS
value: "1"
- name: REPLICATION_FACTOR
value: "1"
- name: DEBEZIUM_REST_HOST_NAME
value: "localhost"
- name: DEBEZIUM_REST_PORT
value: "8083"
- name: DEBEZIUM_PRODUCER_SECURITY_PROTOCOL
value: "SSL"
- name: DEBEZIUM_PRODUCER_SSL_TRUSTSTORE_LOCATION
value: "/mnt/sslcerts/truststore.p12"
- name: DEBEZIUM_PRODUCER_SSL_TRUSTSTORE_PASSWORD
value: {{ .Values.trustStorePassword }}
- name: DEBEZIUM_PRODUCER_SSL_KEYSTORE_LOCATION
value: "/mnt/sslcerts/keystore.p12"
- name: DEBEZIUM_PRODUCER_SSL_KEYSTORE_PASSWORD
value: {{ .Values.keyStorePassword }}
- name: DEBEZIUM_PRODUCER_SSL_KEY_PASSWORD
value: {{ .Values.keyStorePassword }}
- name: DEBEZIUM_CONSUMER_SECURITY_PROTOCOL
value: "SSL"
- name: DEBEZIUM_CONSUMER_SSL_TRUSTSTORE_LOCATION
value: "/mnt/sslcerts/truststore.p12"
- name: DEBEZIUM_CONSUMER_SSL_TRUSTSTORE_PASSWORD
value: {{ .Values.trustStorePassword }}
- name: DEBEZIUM_CONSUMER_SSL_KEYSTORE_LOCATION
value: "/mnt/sslcerts/keystore.p12"
- name: DEBEZIUM_CONSUMER_SSL_KEYSTORE_PASSWORD
value: {{ .Values.keyStorePassword }}
- name: DEBEZIUM_CONSUMER_SSL_KEY_PASSWORD
value: {{ .Values.keyStorePassword }}
image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
ports:
- name: http
containerPort: 80
protocol: TCP
resources:
limits:
memory: 4Gi
cpu: 4
requests:
memory: 1Gi
cpu: 750m
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.tolerations }}
tolerations:
{{- toYaml . | nindent 8 }}
{{- end }}

Error i am getting:
"timestamp":"2024-02-26T05:53:59.573Z","severity":"ERROR","message":"Stopping due to error","logging.googleapis.com/sourceLocation":{"function":"org.apache.kafka.connect.cli.AbstractConnectCli.run"},"logging.googleapis.com/insertId":"349","_exception":{"class":"org.apache.kafka.connect.errors.ConnectException","message":"Failed to connectto and describe Kafka cluster. Check worker's broker connection and security properties.","stackTrace":"org.apache.kafka.connect.errors.ConnectException: Failed to connect to and describe Kafka cluster. Check worker's broker connection and security properties.\n\tat org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:283) ~[connect-runtime-3.5.1.jar:?]\n\tat org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:263) ~[connect-runtime-3.5.1.jar:?]\n\tat org.apache.kafka.connect.runtime.WorkerConfig.kafkaClusterId(WorkerConfig.java:393) ~[connect-runtime-3.5.1.jar:?]\n\tat org.apache.kafka.connect.cli.AbstractConnectCli.startConnect(AbstractConnectCli.java:124) ~[connect-runtime-3.5.1.jar:?]\n\tat org.apache.kafka.connect.cli.AbstractConnectCli.run(AbstractConnectCli.java:94) [connect-runtime-3.5.1.jar:?]\n\tat org.apache.kafka.connect.cli.ConnectDistributed.main(ConnectDistributed.java:116) [connect-runtime-3.5.1.jar:?]\nCaused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listNodes\n\tat java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396) ~[?:?]\n\tat java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073) ~[?:?]\n\tat org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165) ~[kafka-clients-3.5.1.jar:?]\n\tat org.apache.kafka.connect.runtime.WorkerConfig.lookupKafkaClusterId(WorkerConfig.java:277) ~[connect-runtime-3.5.1.jar:?]\n\t... 5 more\nCaused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: listNodes\n"},"_thread":"main","_logger":"org.apache.kafka.connect.cli.AbstractConnectCli"}

[msillence]

the error is: Failed to connectto and describe Kafka cluster. Check worker's broker connection and security

It looks like you've not got any authentication configured so I'd guess connectivity or a problem with your setting of the DEBEZIUM_BOOTSTRAP_SERVERS ?

[ash1912sha]

https://github.com/jhc-systems/debezium-connector-ibmi/blob/main/debezium-connector-ibmi/src/test/resources/confluent-auth-template.properties

shall i give based on the above file like which includes plain user name and password and auth is SASL_SSL ?

or i am right path with certificates to connect to kafka cluster ?

[KanchanJoyia]

Hi Martin,

can you please help on above query

[msillence]

Sorry it's hard to advise, I don't know how the kafka server you are trying to talk to is configured.
Is it expecting authentication is it mutual tls or plain text over ssl or scram?
There are many options i've only used plain text over ssl or none