{"id":405693,"date":"2024-06-29T18:29:13","date_gmt":"2024-06-29T18:29:13","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=405693"},"modified":"-0001-11-30T00:00:00","modified_gmt":"2024-06-29T18:29:13","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=405693","title":{"rendered":"<span>Apache Kafka and AWS S3: backup and restore<\/span>"},"content":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<p>It hasn&#8217;t been that long since my last cheat sheet was published and this is the day for the new one. This time I got a task to find a solution for backing up Kafka to S3 bucket, following the terms:<\/p>\n<ol>\n<li>\n<p>Kafka is running in on-prem k8s cluster;<\/p>\n<\/li>\n<li>\n<p>No managed services;<\/p>\n<\/li>\n<li>\n<p>Use Strimzi Kafka Operator;<\/p>\n<\/li>\n<li>\n<p>All components of the solution are distributing under Apache license;<\/p>\n<\/li>\n<\/ol>\n<p>Sounds cool, right? I think so, too! So what&#8217;s the problem, you ask? As every time before &#8212; no comprehensive manual. All articles are outdated, links are broken and solving the task was kind of solving puzzle. So the only way for me to keep it stuck together, is to write it down and to share with you.<\/p>\n<p>Prerequisites: <\/p>\n<ul>\n<li>\n<p>2 k8s clusters: one for a source Kafka, another one for a target<\/p>\n<\/li>\n<li>\n<p>S3 bucket<\/p>\n<\/li>\n<li>\n<p>Private Docker registry with permissions for pushing and pulling images<\/p>\n<\/li>\n<li>\n<p>Helm installed<\/p>\n<\/li>\n<\/ul>\n<p>The plan:<\/p>\n<ol>\n<li>\n<p>Install Strimzi Kafka Operator and cluster on both k8s clusters<\/p>\n<\/li>\n<li>\n<p>Create topic, send and receive message there<\/p>\n<\/li>\n<li>\n<p>Install Kafka Connect<\/p>\n<\/li>\n<li>\n<p>Install Apache Camel plugin connectors: for <em>S3 sink<\/em> on source Kafka and <em>S3 source<\/em> on target Kafka<\/p>\n<\/li>\n<\/ol>\n<p>Actually, most of the article will be code snippets, so let&#8217;s go to copy-paste!<\/p>\n<h2>Strimzi Kafka Operator and cluster<\/h2>\n<p>Process is straightforward: create namespace, add repo to Helm, install.<\/p>\n<p>Namespace:<\/p>\n<pre><code class=\"bash\">kubectl create ns kafka<\/code><\/pre>\n<p>Add repo to Helm and check it added successfully:<\/p>\n<pre><code class=\"bash\">helm repo add strimzi https:\/\/strimzi.io\/charts\/ helm search repo strimzi<\/code><\/pre>\n<p>Output of the search through repo <em>strimzi<\/em> should be like this:<\/p>\n<pre><code>~$ helm search repo strimzi NAME                            CHART VERSION   APP VERSION     DESCRIPTION strimzi\/strimzi-drain-cleaner   0.4.2           0.4.2           Utility which helps with moving the Apache Kafk... strimzi\/strimzi-kafka-operator  0.35.1          0.35.1          Strimzi: Apache Kafka running on Kubernetes<\/code><\/pre>\n<p>Installing Srimzi Kafka Operator in <em>kafka<\/em> namespace:<\/p>\n<pre><code class=\"bash\">helm install strimzi-kafka-operator strimzi\/strimzi-kafka-operator -n kafka<\/code><\/pre>\n<p>Checking what&#8217;s up after install:<\/p>\n<pre><code class=\"bash\">kubectl get pods -n kafka<\/code><\/pre>\n<p>Output of the command should be like this:<\/p>\n<pre><code>~$ kubectl get pods -n kafka NAME                                       READY   STATUS    RESTARTS   AGE strimzi-cluster-operator-xxxxxxxxx-xxxxx   1\/1     Running   0          1m<\/code><\/pre>\n<p>Operator implements a whole bunch of CRDs, so let&#8217;s utilize some. Create manifest named <em>kafka.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: Kafka metadata:   name: kafka-test-cluster spec:   kafka:     version: 3.4.0     replicas: 1     listeners:       - name: plain         port: 9092         type: internal         tls: false       - name: tls         port: 9093         type: internal         tls: true     config:       offsets.topic.replication.factor: 1       transaction.state.log.replication.factor: 1       transaction.state.log.min.isr: 1       default.replication.factor: 1       min.insync.replicas: 1       inter.broker.protocol.version: \"3.4\"     storage:       type: jbod       volumes:       - id: 0         type: persistent-claim         size: 5Gi         deleteClaim: false   zookeeper:     replicas: 1     storage:       type: persistent-claim       size: 5Gi       deleteClaim: false   entityOperator:     topicOperator: {}     userOperator: {}<\/code><\/pre>\n<p>After you <code>kubectl apply<\/code> this manifest, operator will create 1 replica of ZooKeeper and 1 replica of  Kafka Cluster. Both are with 5Gb persistent storage. Cluster will be named <em>kafka-test-cluster<\/em>:<\/p>\n<pre><code>kubectl apply -f kafka.yaml -n kafka<\/code><\/pre>\n<p>Take a look at the result:<\/p>\n<pre><code>~$ kubectl get pods -n kafka NAME                                       READY   STATUS    RESTARTS   AGE NAME                                                  READY   STATUS    RESTARTS   AGE kafka-test-cluster-entity-operator-xxxxxxxxxx-xxxxx   3\/3     Running   0          115s kafka-test-cluster-kafka-0                            1\/1     Running   0          2m18s kafka-test-cluster-zookeeper-0                        1\/1     Running   0          2m42s strimzi-cluster-operator-xxxxxxxxx-xxxxx              1\/1     Running   0          5m<\/code><\/pre>\n<p>I assume all is up and running, so move next. Let&#8217;s create topic and check message flow.<\/p>\n<h2>Create topic, send and receive message<\/h2>\n<p>Create <em>test-topic.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaTopic metadata:   generation: 1   labels:     strimzi.io\/cluster: kafka-test-cluster   name: ips spec:   config: {}   partitions: 1   replicas: 1<\/code><\/pre>\n<p>And apply it:<\/p>\n<pre><code class=\"bash\">kubectl apply -f test-topic.yaml -n kafka<\/code><\/pre>\n<p>Check if topic is in place:<\/p>\n<pre><code class=\"bash\">kubectl get kafkatopic -n kafka<\/code><\/pre>\n<p>Output be like:<\/p>\n<pre><code>~$ kubectl get kafkatopic -n kafka NAME    CLUSTER                 PARTITIONS   REPLICATION FACTOR   READY test    kafka-test-cluster      1            1                    True<\/code><\/pre>\n<p>Let&#8217;s check how messages flow. But first we need to know Kafka bootstrap&#8217;s IP in the cluster:<\/p>\n<pre><code class=\"bash\">kubectl get svc -n kafka<\/code><\/pre>\n<p>Output:<\/p>\n<pre><code>~$ kubectl get svc -n kafka NAME                                  TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                               AGE kafka-test-cluster-kafka-bootstrap    ClusterIP   10.104.63.229   &lt;none>        9091\/TCP,9092\/TCP,9093\/TCP            15m kafka-test-cluster-kafka-brokers      ClusterIP   None            &lt;none>        9090\/TCP,9091\/TCP,9092\/TCP,9093\/TCP   15m kafka-test-cluster-zookeeper-client   ClusterIP   10.108.31.185   &lt;none>        2181\/TCP                              16m kafka-test-cluster-zookeeper-nodes    ClusterIP   None            &lt;none>        2181\/TCP,2888\/TCP,3888\/TCP            16m kubernetes                            ClusterIP   10.96.0.1       &lt;none>        443\/TCP                               20h<\/code><\/pre>\n<p>Okay, we will aim to <code>10.104.63.229<\/code>. Exec interactive terminal with bash in <em>kafka-test-cluster-kafka-0<\/em> node:<\/p>\n<pre><code>kubectl exec -it -n kafka kafka-test-cluster-kafka-0 -- bash<\/code><\/pre>\n<p>As we are in, change directory to <code>\/opt\/kafka\/bin\/<\/code> and run <code>kafka-console-producer<\/code> script as follows:<\/p>\n<pre><code class=\"bash\">cd \/opt\/kafka\/bin\/ .\/kafka-console-producer.sh --broker-list 10.104.63.229:9092 --topic test<\/code><\/pre>\n<p>Now type in a message and send it by pressing enter:<\/p>\n<pre><code>[kafka@strimzi-kafka-cluster-kafka-0 bin]$ .\/kafka-console-producer.sh --bootstrap-server 10.104.63.229:9092 --topic test >Test footsteps in the darkness of the console! ><\/code><\/pre>\n<p>Let&#8217;s check if it&#8217;s there. Ctrl+C to leave producer&#8217;s console and run consumer&#8217;s one with a command:<\/p>\n<pre><code class=\"bash\">.\/kafka-console-consumer.sh --bootstrap-server 10.104.63.229:9092 --topic test --from-beginning<\/code><\/pre>\n<p>Hold on for a couple of seconds and the result:<\/p>\n<pre><code>[kafka@strimzi-kafka-cluster-kafka-0 bin]$ .\/kafka-console-consumer.sh --bootstrap-server 10.104.63.229:9092 --topic test --from-beginning Test footsteps in the darkness of the console!<\/code><\/pre>\n<h2>Kafka Connect<\/h2>\n<p>Create <em>kafka-connect.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaConnect metadata:   name: kafka-connect-cluster   annotations:     strimzi.io\/use-connector-resources: \"true\" spec:   replicas: 1   authentication:     type: tls     certificateAndKey:       certificate: kafka-test-cluster-kafka-0.crt       key: kafka-test-cluster-kafka-0.key       secretName: kafka-test-cluster-kafka-brokers   bootstrapServers: kafka-test-cluster-kafka-bootstrap:9093   tls:     trustedCertificates:       - secretName: kafka-test-cluster-cluster-ca-cert         certificate: ca.crt   config:     group.id: kafka-connect-cluster     offset.storage.topic: kafka-connect-cluster-offsets     config.storage.topic: kafka-connect-cluster-configs     status.storage.topic: kafka-connect-cluster-status     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.replication.factor: 1     offset.storage.replication.factor: 1     status.storage.replication.factor: 1   build:     output:       type: docker       image: &lt;AWS-ACCOUNT-ID>.dkr.ecr.&lt;AWS-ECR-REGION>.amazonaws.com\/strimzi-kafkaconnect-plugins:s3-sink       pushSecret: ecr-secret     plugins:       - name: camel-aws-s3-sink-kafka-connector         artifacts:           - type: tgz             url: https:\/\/repo.maven.apache.org\/maven2\/org\/apache\/camel\/kafkaconnector\/camel-aws-s3-sink-kafka-connector\/3.18.2\/camel-aws-s3-sink-kafka-connector-3.18.2-package.tar.gz   externalConfiguration:     env:       - name: AWS_ACCESS_KEY_ID         valueFrom:           secretKeyRef:             name: aws-creds             key: AWS_ACCESS_KEY_ID       - name: AWS_SECRET_ACCESS_KEY         valueFrom:           secretKeyRef:             name: aws-creds             key: AWS_SECRET_ACCESS_KEY<\/code><\/pre>\n<p>As it follows from the manifest above we need two secrets to add: one for S3 named <code>aws-creds<\/code> and one for docker registry named <code>ecr-secret<\/code>.<\/p>\n<p>Add secret <code>ecr-secret<\/code> for private registry. In my case it&#8217;s ECR one, located in the region <code>&lt;AWS-ECR-REGION><\/code> of the account <code>&lt;AWS-ACCOUNT-ID><\/code>, so my username is <code>AWS<\/code> and password is generated by <code>aws cli<\/code>:<\/p>\n<pre><code class=\"bash\">kubectl create secret docker-registry ecr-secret \\         --docker-server=&lt;AWS-ACCOUNT-ID>.dkr.ecr.&lt;AWS-ECR-REGION>.amazonaws.com \\ --docker-username=AWS \\ --docker-password=$(aws ecr get-login-password) \\ -n kafka<\/code><\/pre>\n<details class=\"spoiler\">\n<summary>If you don&#8217;t use ECR<\/summary>\n<div class=\"spoiler__content\">\n<p>Change the values of the flags above in accordance with your needs.<\/p>\n<\/div>\n<\/details>\n<p>Add secret <code>aws-creds<\/code>. Base64 encode AWS Access Key and Secret and create <em>aws-creds.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: v1 kind: Secret metadata:   name: aws-creds type: Opaque data:   AWS_ACCESS_KEY_ID: &lt;Base64-encoded-Key>   AWS_SECRET_ACCESS_KEY: &lt;Base64-encoded-Secret><\/code><\/pre>\n<p>Now apply <em>aws-creds.yaml<\/em> and check if all the secrets are in place:<\/p>\n<pre><code class=\"bash\">kubectl apply -f aws-creds.yaml -n kafka kubectl get secret -n kafka<\/code><\/pre>\n<p>Output be like:<\/p>\n<pre><code>~$ kubectl get secret -n kafka NAME                                                TYPE                             DATA   AGE aws-creds                                           Opaque                           2      2m ecr-secret                                          kubernetes.io\/dockerconfigjson   1      3m<\/code><\/pre>\n<p>Now it&#8217;s time for Kafka Connect itself. Apply <em>kafka-connect.yaml<\/em>:<\/p>\n<pre><code>kubectl apply -f kafka-connect.yaml -n kafka<\/code><\/pre>\n<p>Let&#8217;s check how that works:<\/p>\n<pre><code>kubectl get pods -n kafka<\/code><\/pre>\n<pre><code>~$ kubectl get pods NAME                                                  READY   STATUS    RESTARTS   AGE kafka-connect-cluster-connect-build                   1\/1     Running   0          77s kafka-test-cluster-entity-operator-xxxxxxxxxx-xxxxx   3\/3     Running   0          70m kafka-test-cluster-kafka-0                            1\/1     Running   0          70m kafka-test-cluster-zookeeper-0                        1\/1     Running   0          71m strimzi-cluster-operator-xxxxxxxxx-xxxxx              1\/1     Running   0          21h<\/code><\/pre>\n<p>As we can see, Kafka Connect started pod <code>kafka-connect-cluster-connect-build<\/code> which builds image with configuration we set up.<\/p>\n<p>After a while checking again:<\/p>\n<pre><code>~$ kubectl get pods NAME                                                  READY   STATUS             RESTARTS   AGE kafka-connect-cluster-connect-xxxxxxxxx-xxxxx         1\/1     Running            0          100s kafka-test-cluster-entity-operator-xxxxxxxxxx-xxxxx   3\/3     Running            0          72m kafka-test-cluster-kafka-0                            1\/1     Running            0          72m kafka-test-cluster-zookeeper-0                        1\/1     Running            0          73m strimzi-cluster-operator-xxxxxxxxx-xxxxx              1\/1     Running            0          21h<\/code><\/pre>\n<details class=\"spoiler\">\n<summary>What if STATUS of the pod is ImagePullBackOff<\/summary>\n<div class=\"spoiler__content\">\n<p>For reasons which are not clear for me, resource<code>KafkaConnect<\/code> doesn&#8217;t put <code>imagePullSecrets<\/code> into <code>spec<\/code> section of deployment of the kind <code>strimzi.io\/kind: KafkaConnect.<\/code><\/p>\n<p>You can fix it easily by editing the deployment in-place:<\/p>\n<pre><code>kubectl edit deploy kafka-connect-cluster-connect -n kafka<\/code><\/pre>\n<p>And adding <code>imagePullSecrets <\/code>in<code> spec <\/code>section right between<code> affinity <\/code>and<code> containers <\/code>(and not to forget about correct indentation):<\/p>\n<p>Save and exit. <\/p>\n<p>I may also patch deployment in-place:<\/p>\n<pre><code class=\"bash\">kubectl patch kafka-connect-cluster-connect -p '{\"spec\":{\"template\":{\"spec\":{\"imagePullSecrets\":[{\"name\":\"ecr-secret\"}]}}}}'<\/code><\/pre>\n<p>Check the result:<\/p>\n<pre><code>ubuntu@ip-10-200-50-148:~$ kubectl get pods NAME                                                  READY   STATUS              RESTARTS   AGE kafka-connect-cluster-connect-yyyyyyyyy-yyyyyy        0\/1     ContainerCreating   0          3s kafka-connect-cluster-connect-xxxxxxxxx-xxxxx         0\/1     ImagePullBackOff    0          19m kafka-test-cluster-entity-operator-xxxxxxxxxx-xxxxx   3\/3     Running             0          90m kafka-test-cluster-kafka-0                            1\/1     Running             0          90m kafka-test-cluster-zookeeper-0                        1\/1     Running             0          90m strimzi-cluster-operator-xxxxxxxxx-xxxxx              1\/1     Running             0          21h<\/code><\/pre>\n<p>Container is successfully pulled<\/p>\n<\/div>\n<\/details>\n<p>Okay. That&#8217;s fun, so let&#8217;s double it and repeat all the steps above for the target cluster. \ud83d\ude00<\/p>\n<p>When you&#8217;re ready with the target one, it&#8217;s a time for connectors and actual backup and restore.<\/p>\n<h2>Install Apache Camel plugin connectors<\/h2>\n<p>Create <em>s3-sink.yaml<\/em> for the <strong>source<\/strong> cluster:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaConnector metadata:   name: s3-sink-connector   labels:     strimzi.io\/cluster: kafka-connect-cluster spec:   class: org.apache.camel.kafkaconnector.awss3sink.CamelAwss3sinkSinkConnector   tasksMax: 1   config:     key.converter: org.apache.kafka.connect.storage.StringConverter     value.converter: org.apache.kafka.connect.storage.StringConverter     topics: ips     camel.kamelet.aws-s3-sink.bucketNameOrArn: \"arn:aws:s3:::ARN-OF-YOUR-BUCKET\"     camel.kamelet.aws-s3-sink.accessKey: \"&lt;AWS-ACCESS-KEY>\"     camel.kamelet.aws-s3-sink.secretKey: \"&lt;AWS-SECRET-KEY>\"     camel.kamelet.aws-s3-sink.region: \"&lt;AWS-S3-REGION>\"     camel.kamelet.aws-s3-sink.keyName: \"test\"<\/code><\/pre>\n<blockquote>\n<p>Please notice values for <code>spec.config.key.converter<\/code> and <code>spec.config.value.converter<\/code>. For the sake of simplicity in the article I want to save messages as plain-text objects in S3 bucket and this is why is use <code>org.apache.kafka.connect.storage.StringConverter<\/code> here.<\/p>\n<\/blockquote>\n<p>Apply yaml and send message to the test topic as we did it before:<\/p>\n<pre><code>kubectl apply -f s3-sink.yaml -n kafka<\/code><\/pre>\n<p>Exec bash in interactive terminal again and send message which we are going to catch on the other side:<\/p>\n<pre><code class=\"bash\">kubectl exec -it -n kafka kafka-test-cluster-kafka-0 -- bash cd \/opt\/kafka\/bin\/ .\/kafka-console-producer.sh --broker-list 10.104.63.229:9092 --topic test<\/code><\/pre>\n<pre><code>[kafka@strimzi-kafka-cluster-kafka-0 bin]$ .\/kafka-console-producer.sh --bootstrap-server 10.104.63.229:9092 --topic test >Wait for me and I shall come! ><\/code><\/pre>\n<p>Let&#8217;s take a look at the bucket:<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/55f\/462\/28b\/55f46228b73a7daa43fd74a147cac4e8.png\" width=\"1549\" height=\"365\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/55f\/462\/28b\/55f46228b73a7daa43fd74a147cac4e8.png\"\/><\/figure>\n<p>Object&#8217;s in place.<\/p>\n<p>Let&#8217;s take a look what&#8217;s inside:<\/p>\n<figure class=\"\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/765\/ca3\/857\/765ca3857c04fdbbd931141c5d4ab87b.png\" width=\"465\" height=\"281\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/765\/ca3\/857\/765ca3857c04fdbbd931141c5d4ab87b.png\"\/><\/figure>\n<p>Okay. Backup is working.<\/p>\n<p>Let&#8217;s set up restore.<\/p>\n<p>Create <em>s3-source.yaml<\/em> for the <strong>target<\/strong> cluster:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaConnector metadata:   name: s3-source-connector   labels:     strimzi.io\/cluster: kafka-connect-cluster spec:   class: org.apache.camel.kafkaconnector.awss3source.CamelAwss3sourceSourceConnector   tasksMax: 1   config:     key.converter: org.apache.kafka.connect.storage.StringConverter     value.converter: org.apache.kafka.connect.converters.ByteArrayConverter     topics: test     camel.kamelet.aws-s3-source.bucketNameOrArn: \"arn:aws:s3:::ARN-OF-YOUR-BUCKET\"     camel.kamelet.aws-s3-source.deleteAfterRead: \"false\"     camel.kamelet.aws-s3-source.accessKey: \"&lt;AWS-ACCESS-KEY>\"     camel.kamelet.aws-s3-source.secretKey: \"&lt;AWS-SECRET-KEY>\"     camel.kamelet.aws-s3-source.region: \"&lt;AWS-S3-REGION>\"<\/code><\/pre>\n<blockquote>\n<p>And again, please notice values for <code>spec.config.key.converter<\/code> and <code>spec.config.value.converter<\/code>. Since connector gets byte stream from S3 we need to use  <code>org.apache.kafka.connect.converters.ByteArrayConverter<\/code>.<\/p>\n<p>Also please take a look at the parameter <code>camel.kamelet.aws-s3-source.deleteAfterRead<\/code> and it&#8217;s value <code>false<\/code>. If you change it to <code>true<\/code> you&#8217;ll lose all your backed up messages as soon as connector read them. <\/p>\n<\/blockquote>\n<p>Apply <em>s3-source.yaml <\/em>and let&#8217;s see what we received:<\/p>\n<pre><code class=\"bash\">kubectl apply -f s3-source.yaml -n kafka<\/code><\/pre>\n<p>Exec bash in the interactive terminal again and run consumer:<\/p>\n<pre><code class=\"bash\">kubectl exec -it -n kafka kafka-test-cluster-kafka-0 -- bash cd \/opt\/kafka\/bin\/ .\/kafka-console-consumer.sh --bootstrap-server 10.87.163.29:9092 --topic test --from-beginning<\/code><\/pre>\n<p>And the output:<\/p>\n<pre><code>[kafka@kafka-test-cluster-kafka-0 bin]$ .\/kafka-console-consumer.sh --bootstrap-server 10.87.163.29:9092 --topic test --from-beginning Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come! Wait for me and I shall come!<\/code><\/pre>\n<p>Message has come. <\/p>\n<p>But why so many times, you ask? Because of <code>camel.kamelet.aws-s3-source.deleteAfterRead: \"false\" <\/code>and <code>camel.kamelet.aws-s3-source.delay<\/code> parameter which we were not set and which default value is 500ms.<\/p>\n<p>I hope this article will ease you start in setting up Kafka backup and restore. Happy playing!<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/7f2\/a74\/70e\/7f2a7470edea1acc9fd0e2a023451933.png\" width=\"1024\" height=\"587\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/7f2\/a74\/70e\/7f2a7470edea1acc9fd0e2a023451933.png\"\/><\/figure>\n<p>List of used resources:<\/p>\n<ul>\n<li>\n<p><a href=\"https:\/\/strimzi.io\/blog\/2018\/11\/01\/using-helm\/\" rel=\"noopener noreferrer nofollow\">https:\/\/strimzi.io\/blog\/2018\/11\/01\/using-helm\/<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/strimzi.io\/docs\/operators\/latest\/configuring.html#proc-kafka-connect-config-str\" rel=\"noopener noreferrer nofollow\">https:\/\/strimzi.io\/docs\/operators\/latest\/configuring.html#proc-kafka-connect-config-str<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/camel.apache.org\/camel-kafka-connector\/3.18.x\/reference\/connectors\/camel-aws-s3-sink-kafka-sink-connector.html\" rel=\"noopener noreferrer nofollow\">https:\/\/camel.apache.org\/camel-kafka-connector\/3.18.x\/reference\/connectors\/camel-aws-s3-sink-kafka-sink-connector.html<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/camel.apache.org\/camel-kafka-connector\/3.18.x\/reference\/connectors\/camel-aws-s3-source-kafka-source-connector.html#_connector_option_camel_kamelet_aws-s3-source_bucketNameOrArn\" rel=\"noopener noreferrer nofollow\">https:\/\/camel.apache.org\/camel-kafka-connector\/3.18.x\/reference\/connectors\/camel-aws-s3-source-kafka-source-connector.html<\/a><\/p>\n<\/li>\n<\/ul>\n<\/div>\n<\/div>\n<\/div>\n<p><!----><!----><\/div>\n<p><!----><!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/articles\/744790\/\"> https:\/\/habr.com\/ru\/articles\/744790\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><!--[--><!--]--><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<p>It hasn&#8217;t been that long since my last cheat sheet was published and this is the day for the new one. This time I got a task to find a solution for backing up Kafka to S3 bucket, following the terms:<\/p>\n<ol>\n<li>\n<p>Kafka is running in on-prem k8s cluster;<\/p>\n<\/li>\n<li>\n<p>No managed services;<\/p>\n<\/li>\n<li>\n<p>Use Strimzi Kafka Operator;<\/p>\n<\/li>\n<li>\n<p>All components of the solution are distributing under Apache license;<\/p>\n<\/li>\n<\/ol>\n<p>Sounds cool, right? I think so, too! So what&#8217;s the problem, you ask? As every time before &#8212; no comprehensive manual. All articles are outdated, links are broken and solving the task was kind of solving puzzle. So the only way for me to keep it stuck together, is to write it down and to share with you.<\/p>\n<p>Prerequisites: <\/p>\n<ul>\n<li>\n<p>2 k8s clusters: one for a source Kafka, another one for a target<\/p>\n<\/li>\n<li>\n<p>S3 bucket<\/p>\n<\/li>\n<li>\n<p>Private Docker registry with permissions for pushing and pulling images<\/p>\n<\/li>\n<li>\n<p>Helm installed<\/p>\n<\/li>\n<\/ul>\n<p>The plan:<\/p>\n<ol>\n<li>\n<p>Install Strimzi Kafka Operator and cluster on both k8s clusters<\/p>\n<\/li>\n<li>\n<p>Create topic, send and receive message there<\/p>\n<\/li>\n<li>\n<p>Install Kafka Connect<\/p>\n<\/li>\n<li>\n<p>Install Apache Camel plugin connectors: for <em>S3 sink<\/em> on source Kafka and <em>S3 source<\/em> on target Kafka<\/p>\n<\/li>\n<\/ol>\n<p>Actually, most of the article will be code snippets, so let&#8217;s go to copy-paste!<\/p>\n<h2>Strimzi Kafka Operator and cluster<\/h2>\n<p>Process is straightforward: create namespace, add repo to Helm, install.<\/p>\n<p>Namespace:<\/p>\n<pre><code class=\"bash\">kubectl create ns kafka<\/code><\/pre>\n<p>Add repo to Helm and check it added successfully:<\/p>\n<pre><code class=\"bash\">helm repo add strimzi https:\/\/strimzi.io\/charts\/ helm search repo strimzi<\/code><\/pre>\n<p>Output of the search through repo <em>strimzi<\/em> should be like this:<\/p>\n<pre><code>~$ helm search repo strimzi NAME                            CHART VERSION   APP VERSION     DESCRIPTION strimzi\/strimzi-drain-cleaner   0.4.2           0.4.2           Utility which helps with moving the Apache Kafk... strimzi\/strimzi-kafka-operator  0.35.1          0.35.1          Strimzi: Apache Kafka running on Kubernetes<\/code><\/pre>\n<p>Installing Srimzi Kafka Operator in <em>kafka<\/em> namespace:<\/p>\n<pre><code class=\"bash\">helm install strimzi-kafka-operator strimzi\/strimzi-kafka-operator -n kafka<\/code><\/pre>\n<p>Checking what&#8217;s up after install:<\/p>\n<pre><code class=\"bash\">kubectl get pods -n kafka<\/code><\/pre>\n<p>Output of the command should be like this:<\/p>\n<pre><code>~$ kubectl get pods -n kafka NAME                                       READY   STATUS    RESTARTS   AGE strimzi-cluster-operator-xxxxxxxxx-xxxxx   1\/1     Running   0          1m<\/code><\/pre>\n<p>Operator implements a whole bunch of CRDs, so let&#8217;s utilize some. Create manifest named <em>kafka.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: Kafka metadata:   name: kafka-test-cluster spec:   kafka:     version: 3.4.0     replicas: 1     listeners:       - name: plain         port: 9092         type: internal         tls: false       - name: tls         port: 9093         type: internal         tls: true     config:       offsets.topic.replication.factor: 1       transaction.state.log.replication.factor: 1       transaction.state.log.min.isr: 1       default.replication.factor: 1       min.insync.replicas: 1       inter.broker.protocol.version: \"3.4\"     storage:       type: jbod       volumes:       - id: 0         type: persistent-claim         size: 5Gi         deleteClaim: false   zookeeper:     replicas: 1     storage:       type: persistent-claim       size: 5Gi       deleteClaim: false   entityOperator:     topicOperator: {}     userOperator: {}<\/code><\/pre>\n<p>After you <code>kubectl apply<\/code> this manifest, operator will create 1 replica of ZooKeeper and 1 replica of  Kafka Cluster. Both are with 5Gb persistent storage. Cluster will be named <em>kafka-test-cluster<\/em>:<\/p>\n<pre><code>kubectl apply -f kafka.yaml -n kafka<\/code><\/pre>\n<p>Take a look at the result:<\/p>\n<pre><code>~$ kubectl get pods -n kafka NAME                                       READY   STATUS    RESTARTS   AGE NAME                                                  READY   STATUS    RESTARTS   AGE kafka-test-cluster-entity-operator-xxxxxxxxxx-xxxxx   3\/3     Running   0          115s kafka-test-cluster-kafka-0                            1\/1     Running   0          2m18s kafka-test-cluster-zookeeper-0                        1\/1     Running   0          2m42s strimzi-cluster-operator-xxxxxxxxx-xxxxx              1\/1     Running   0          5m<\/code><\/pre>\n<p>I assume all is up and running, so move next. Let&#8217;s create topic and check message flow.<\/p>\n<h2>Create topic, send and receive message<\/h2>\n<p>Create <em>test-topic.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaTopic metadata:   generation: 1   labels:     strimzi.io\/cluster: kafka-test-cluster   name: ips spec:   config: {}   partitions: 1   replicas: 1<\/code><\/pre>\n<p>And apply it:<\/p>\n<pre><code class=\"bash\">kubectl apply -f test-topic.yaml -n kafka<\/code><\/pre>\n<p>Check if topic is in place:<\/p>\n<pre><code class=\"bash\">kubectl get kafkatopic -n kafka<\/code><\/pre>\n<p>Output be like:<\/p>\n<pre><code>~$ kubectl get kafkatopic -n kafka NAME    CLUSTER                 PARTITIONS   REPLICATION FACTOR   READY test    kafka-test-cluster      1            1                    True<\/code><\/pre>\n<p>Let&#8217;s check how messages flow. But first we need to know Kafka bootstrap&#8217;s IP in the cluster:<\/p>\n<pre><code class=\"bash\">kubectl get svc -n kafka<\/code><\/pre>\n<p>Output:<\/p>\n<pre><code>~$ kubectl get svc -n kafka NAME                                  TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                               AGE kafka-test-cluster-kafka-bootstrap    ClusterIP   10.104.63.229   &lt;none>        9091\/TCP,9092\/TCP,9093\/TCP            15m kafka-test-cluster-kafka-brokers      ClusterIP   None            &lt;none>        9090\/TCP,9091\/TCP,9092\/TCP,9093\/TCP   15m kafka-test-cluster-zookeeper-client   ClusterIP   10.108.31.185   &lt;none>        2181\/TCP                              16m kafka-test-cluster-zookeeper-nodes    ClusterIP   None            &lt;none>        2181\/TCP,2888\/TCP,3888\/TCP            16m kubernetes                            ClusterIP   10.96.0.1       &lt;none>        443\/TCP                               20h<\/code><\/pre>\n<p>Okay, we will aim to <code>10.104.63.229<\/code>. Exec interactive terminal with bash in <em>kafka-test-cluster-kafka-0<\/em> node:<\/p>\n<pre><code>kubectl exec -it -n kafka kafka-test-cluster-kafka-0 -- bash<\/code><\/pre>\n<p>As we are in, change directory to <code>\/opt\/kafka\/bin\/<\/code> and run <code>kafka-console-producer<\/code> script as follows:<\/p>\n<pre><code class=\"bash\">cd \/opt\/kafka\/bin\/ .\/kafka-console-producer.sh --broker-list 10.104.63.229:9092 --topic test<\/code><\/pre>\n<p>Now type in a message and send it by pressing enter:<\/p>\n<pre><code>[kafka@strimzi-kafka-cluster-kafka-0 bin]$ .\/kafka-console-producer.sh --bootstrap-server 10.104.63.229:9092 --topic test >Test footsteps in the darkness of the console! ><\/code><\/pre>\n<p>Let&#8217;s check if it&#8217;s there. Ctrl+C to leave producer&#8217;s console and run consumer&#8217;s one with a command:<\/p>\n<pre><code class=\"bash\">.\/kafka-console-consumer.sh --bootstrap-server 10.104.63.229:9092 --topic test --from-beginning<\/code><\/pre>\n<p>Hold on for a couple of seconds and the result:<\/p>\n<pre><code>[kafka@strimzi-kafka-cluster-kafka-0 bin]$ .\/kafka-console-consumer.sh --bootstrap-server 10.104.63.229:9092 --topic test --from-beginning Test footsteps in the darkness of the console!<\/code><\/pre>\n<h2>Kafka Connect<\/h2>\n<p>Create <em>kafka-connect.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: kafka.strimzi.io\/v1beta2 kind: KafkaConnect metadata:   name: kafka-connect-cluster   annotations:     strimzi.io\/use-connector-resources: \"true\" spec:   replicas: 1   authentication:     type: tls     certificateAndKey:       certificate: kafka-test-cluster-kafka-0.crt       key: kafka-test-cluster-kafka-0.key       secretName: kafka-test-cluster-kafka-brokers   bootstrapServers: kafka-test-cluster-kafka-bootstrap:9093   tls:     trustedCertificates:       - secretName: kafka-test-cluster-cluster-ca-cert         certificate: ca.crt   config:     group.id: kafka-connect-cluster     offset.storage.topic: kafka-connect-cluster-offsets     config.storage.topic: kafka-connect-cluster-configs     status.storage.topic: kafka-connect-cluster-status     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.replication.factor: 1     offset.storage.replication.factor: 1     status.storage.replication.factor: 1   build:     output:       type: docker       image: &lt;AWS-ACCOUNT-ID>.dkr.ecr.&lt;AWS-ECR-REGION>.amazonaws.com\/strimzi-kafkaconnect-plugins:s3-sink       pushSecret: ecr-secret     plugins:       - name: camel-aws-s3-sink-kafka-connector         artifacts:           - type: tgz             url: https:\/\/repo.maven.apache.org\/maven2\/org\/apache\/camel\/kafkaconnector\/camel-aws-s3-sink-kafka-connector\/3.18.2\/camel-aws-s3-sink-kafka-connector-3.18.2-package.tar.gz   externalConfiguration:     env:       - name: AWS_ACCESS_KEY_ID         valueFrom:           secretKeyRef:             name: aws-creds             key: AWS_ACCESS_KEY_ID       - name: AWS_SECRET_ACCESS_KEY         valueFrom:           secretKeyRef:             name: aws-creds             key: AWS_SECRET_ACCESS_KEY<\/code><\/pre>\n<p>As it follows from the manifest above we need two secrets to add: one for S3 named <code>aws-creds<\/code> and one for docker registry named <code>ecr-secret<\/code>.<\/p>\n<p>Add secret <code>ecr-secret<\/code> for private registry. In my case it&#8217;s ECR one, located in the region <code>&lt;AWS-ECR-REGION><\/code> of the account <code>&lt;AWS-ACCOUNT-ID><\/code>, so my username is <code>AWS<\/code> and password is generated by <code>aws cli<\/code>:<\/p>\n<pre><code class=\"bash\">kubectl create secret docker-registry ecr-secret \\         --docker-server=&lt;AWS-ACCOUNT-ID>.dkr.ecr.&lt;AWS-ECR-REGION>.amazonaws.com \\ --docker-username=AWS \\ --docker-password=$(aws ecr get-login-password) \\ -n kafka<\/code><\/pre>\n<details class=\"spoiler\">\n<summary>If you don&#8217;t use ECR<\/summary>\n<div class=\"spoiler__content\">\n<p>Change the values of the flags above in accordance with your needs.<\/p>\n<\/div>\n<\/details>\n<p>Add secret <code>aws-creds<\/code>. Base64 encode AWS Access Key and Secret and create <em>aws-creds.yaml<\/em>:<\/p>\n<pre><code class=\"yaml\">apiVersion: v1 kind: Secret metadata:   name: aws-creds type: Opaque data:   AWS_ACCESS_KEY_ID: &lt;Base64-encoded-Key>   AWS_SECRET_ACCESS_KEY: &lt;Base64-encoded-Secret><\/code><\/pre>\n<p>Now apply <em>aws-creds.yaml<\/em> and check if all the secrets are in place:<\/p>\n<pre><code class=\"bash\">kubectl apply -f aws-creds.yaml -n kafka kubectl get secret -n kafka<\/code><\/pre>\n<p>Output be like:<\/p>\n<pre><code>~$ kubectl get secret -n kafka NAME                                                TYPE                             DATA   AGE aws-creds                                           Opaque                           2      2m ecr-secret                                          kubernetes.io\/dockerconfigjson   1      3m<\/code><\/pre>\n<p>Now it&#8217;s time for Kafka Connect itself. Apply <em>kafka-connect.yaml<\/em>:<\/p>\n<pre><code>kubectl apply -f kafka-connect.yaml -n kafka<\/code><\/pre>\n<p>Let&#8217;s check how that works:<\/p>\n<pre><code>kubectl get pods -n kafka<\/code><\/pre>\n<pre><code>~$ kubectl get pods NAME                           <\/code><\/pre>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-405693","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/405693","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=405693"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/405693\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=405693"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=405693"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=405693"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}