
<p><strong>Which chart</strong>:
charts/bitnami/kafka/
<a href="https://github.com/bitnami/charts/tree/master/bitnami/kafka" rel="nofollow noreferrer">https://github.com/bitnami/charts/tree/master/bitnami/kafka</a></p>

<p><strong>Describe the bug</strong>
I am following the tutorial  <a href="https://docs.bitnami.com/tutorials/build-messaging-cluster-apache-kafka-mongodb-kubernetes/" rel="nofollow noreferrer">Build a Scalable, Fault-Tolerant Messaging Cluster on Kubernetes with Apache Kafka and MongoDB</a></p>
<p>To resolve the extraDeploy issue i followed <a href="https://github.com/bitnami/charts/issues/5649" rel="nofollow noreferrer">extraDeploy doesn't render with example in documentation #5649</a>. The issue is resolved and i have working configuration as below:</p>
<ol>
<li>Dockerfile</li>
</ol>
<pre><code>FROM bitnami/kafka:latest

RUN mkdir -p /opt/bitnami/kafka/plugins &amp;&amp; \
    cd /opt/bitnami/kafka/plugins &amp;&amp; \
    curl --remote-name --location --silent https://search.maven.org/remotecontent?filepath=org/mongodb/kafka/mongo-kafka-connect/1.6.1/mongo-kafka-connect-1.6.1-all.jar
</code></pre>
<ol start="2">
<li>Values.yml</li>
</ol>
<pre><code>extraDeploy:
  - |
    apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: {{ include &quot;kafka.fullname&quot; . }}-connect
      labels: {{- include &quot;common.labels.standard&quot; . | nindent 4 }}
        app.kubernetes.io/component: connector
    spec:
      replicas: 1
      selector:
        matchLabels: {{- include &quot;common.labels.matchLabels&quot; . | nindent 6 }}
          app.kubernetes.io/component: connector
      template:
        metadata:
          labels: {{- include &quot;common.labels.standard&quot; . | nindent 8 }}
            app.kubernetes.io/component: connector
        spec:
          containers:
            - name: connect
              image: kafka-connect-bitnami:5.22
              imagePullPolicy: Never
              command: [ &quot;/bin/bash&quot;, &quot;-c&quot;, &quot;/opt/bitnami/kafka/bin/connect-standalone.sh /config/connect-standalone.properties /config/mongodb-sink.properties&quot;]
              ports:
                - name: connector
                  containerPort: 8083
              volumeMounts:
                - name: configuration
                  mountPath: /config
          volumes:
            - name: configuration
              configMap:
                name: {{ include &quot;kafka.fullname&quot; . }}-connect
  - |
    apiVersion: v1
    kind: ConfigMap
    metadata:
      name: {{ include &quot;kafka.fullname&quot; . }}-connect
      labels: {{- include &quot;common.labels.standard&quot; . | nindent 4 }}
        app.kubernetes.io/component: connector
    data:
      connect-standalone.properties: |-
        bootstrap.servers={{ include &quot;kafka.fullname&quot; . }}.{{ .Release.Namespace }}.svc.{{ .Values.clusterDomain }}:{{ .Values.service.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
        offset.storage.file.filename=/tmp/connect.offsets
        offset.flush.interval.ms=20000
        plugin.path=/opt/bitnami/kafka/plugins
      mongodb-source.properties: |-
        connection.uri=mongodb://user:password@mongodb.default.svc.cluster.local:27017/mydb
        name=mongo-source-connector
        topics=source-topic
        connector.class=com.mongodb.kafka.connect.MongoSourceConnector
        tasks.max=1
        key.converter=org.apache.kafka.connect.json.JsonConverter
        value.converter=org.apache.kafka.connect.json.JsonConverter
        key.converter.schemas.enable=false
        value.converter.schemas.enable=false
        database=mydb
        collection=source
        batch.size=0
        change.stream.full.document=updateLookup
        pipeline=[]
        collation=
      mongodb-sink.properties: |-
        connection.uri=mongodb://user:password@mongodb.default.svc.cluster.local:27017/mydb
        name=mongo-sink-connector
        topics=sink-topic
        connector.class=com.mongodb.kafka.connect.MongoSinkConnector
        tasks.max=1
        key.converter=org.apache.kafka.connect.json.JsonConverter
        value.converter=org.apache.kafka.connect.json.JsonConverter
        key.converter.schemas.enable=false
        value.converter.schemas.enable=false
        database=mydb
        collection=sink
  - |
    apiVersion: v1
    kind: Service    
    metadata:
      name: {{ include &quot;kafka.fullname&quot; . }}-connect
      labels: {{- include &quot;common.labels.standard&quot; . | nindent 4 }}
        app.kubernetes.io/component: connector
    spec:
      ports:
        - protocol: TCP
          port: 8083
          targetPort: connector
      selector: {{- include &quot;common.labels.matchLabels&quot; . | nindent 4 }}
        app.kubernetes.io/component: connector
</code></pre>

<ol start="3">
<li>I have deployed the mongodb with replica using following tutorial [Horizontally Scale Your MongoDB Deployment with Bitnami, Kubernetes and Helm] (<a href="https://engineering.bitnami.com/articles/horizontally-scale-your-mongodb-deployment-with-bitnami-kubernetes-and-helm.html" rel="nofollow noreferrer">https://engineering.bitnami.com/articles/horizontally-scale-your-mongodb-deployment-with-bitnami-kubernetes-and-helm.html</a>)</li>
</ol>
<p>and then i run</p>
<pre><code>helm install kafka bitnami/kafka -f values.yml
</code></pre>
<p>I have following setup up and running for mongodb-sink-connector:</p>
<p><img src="https://user-images.githubusercontent.com/32441119/150311189-dd28d3ce-ff19-4f9d-8e6e-68ea90fae060.png" alt="kafka-mongo-setup" /></p>
<pre><code>NAME                                READY   STATUS    RESTARTS      AGE
pod/kafka-0                         1/1     Running   1 (35h ago)   35h
pod/kafka-client                    1/1     Running   0             43h
pod/kafka-connect-669487944-gb4p7   1/1     Running   0             57m
pod/kafka-zookeeper-0               1/1     Running   0             35h
pod/mongodb-arbiter-0               1/1     Running   0             42h
pod/mongodb-client                  1/1     Running   0             41h
pod/mongodb-primary-0               2/2     Running   0             42h
pod/mongodb-secondary-0             2/2     Running   0             42h

NAME                               TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                      AGE
service/kafka                      ClusterIP   10.98.236.199   &lt;none&gt;        9092/TCP                     35h
service/kafka-connect              ClusterIP   10.105.58.215   &lt;none&gt;        8083/TCP                     35h
service/kafka-headless             ClusterIP   None            &lt;none&gt;        9092/TCP,9093/TCP            35h
service/kafka-zookeeper            ClusterIP   10.108.22.188   &lt;none&gt;        2181/TCP,2888/TCP,3888/TCP   35h
service/kafka-zookeeper-headless   ClusterIP   None            &lt;none&gt;        2181/TCP,2888/TCP,3888/TCP   35h
service/kubernetes                 ClusterIP   10.96.0.1       &lt;none&gt;        443/TCP                      14d
service/mongodb                    ClusterIP   10.96.105.153   &lt;none&gt;        27017/TCP,9216/TCP           42h
service/mongodb-headless           ClusterIP   None            &lt;none&gt;        27017/TCP                    42h

NAME                            READY   UP-TO-DATE   AVAILABLE   AGE
deployment.apps/kafka-connect   1/1     1            1           35h

NAME                                       DESIRED   CURRENT   READY   AGE
replicaset.apps/kafka-connect-54cff6f879   0         0         0       35h
replicaset.apps/kafka-connect-59fcf7754c   0         0         0       24h
replicaset.apps/kafka-connect-64c5697f54   0         0         0       21h
replicaset.apps/kafka-connect-669487944    1         1         1       21h
replicaset.apps/kafka-connect-66c6dd4679   0         0         0       35h
replicaset.apps/kafka-connect-84ffbffd5c   0         0         0       23h

NAME                                 READY   AGE
statefulset.apps/kafka               1/1     35h
statefulset.apps/kafka-zookeeper     1/1     35h
statefulset.apps/mongodb-arbiter     1/1     42h
statefulset.apps/mongodb-primary     1/1     42h
statefulset.apps/mongodb-secondary   1/1     42h

</code></pre>
<p>I can publish the messages to mongodb successfully with above configuration. Below are some message after successful sink on kafaka-connect pod.</p>
<pre><code>[2022-01-19 11:52:25,476] INFO [mongo-sink-connector|task-0] [Consumer clientId=connector-consumer-mongo-sink-connector-0, groupId=connect-mongo-sink-connector] Setting offset for partition sink-topic-0 to the committed offset FetchPosition{offset=3, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[kafka-0.kafka-headless.default.svc.cluster.local:9092 (id: 0 rack: null)], epoch=0}} (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator:818)
[2022-01-19 11:52:25,519] INFO [mongo-sink-connector|task-0] Cluster created with settings {hosts=[mongodb.default.svc.cluster.local:27017], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms'} (org.mongodb.driver.cluster:71)
[2022-01-19 11:52:25,522] INFO [mongo-sink-connector|task-0] Opened connection [connectionId{localValue:5, serverValue:7744}] to mongodb.default.svc.cluster.local:27017 (org.mongodb.driver.connection:71)
[2022-01-19 11:52:25,522] INFO [mongo-sink-connector|task-0] Opened connection [connectionId{localValue:4, serverValue:7743}] to mongodb.default.svc.cluster.local:27017 (org.mongodb.driver.connection:71)
[2022-01-19 11:52:25,523] INFO [mongo-sink-connector|task-0] Monitor thread successfully connected to server with description ServerDescription{address=mongodb.default.svc.cluster.local:27017, type=REPLICA_SET_PRIMARY, state=CONNECTED, ok=true, minWireVersion=0, maxWireVersion=8, maxDocumentSize=16777216, logicalSessionTimeoutMinutes=30, roundTripTimeNanos=2650238, setName='rs0', canonicalAddress=mongodb-primary-0.mongodb-headless.default.svc.cluster.local:27017, hosts=[mongodb-primary-0.mongodb-headless.default.svc.cluster.local:27017, mongodb-secondary-0.mongodb-headless.default.svc.cluster.local:27017], passives=[], arbiters=[mongodb-arbiter-0.mongodb-headless.default.svc.cluster.local:27017], primary='mongodb-primary-0.mongodb-headless.default.svc.cluster.local:27017', tagSet=TagSet{[]}, electionId=7fffffff0000000000000002, setVersion=3, topologyVersion=null, lastWriteDate=Wed Jan 19 11:52:24 UTC 2022, lastUpdateTimeNanos=188123424943131} (org.mongodb.driver.cluster:71)
[2022-01-19 11:52:25,566] INFO [mongo-sink-connector|task-0] Opened connection [connectionId{localValue:6, serverValue:7745}] to mongodb.default.svc.cluster.local:27017 (org.mongodb.driver.connection:71)
</code></pre>
<p><strong>To Reproduce</strong></p>
<p>From producer:</p>
<pre><code>kubectl exec --tty -i kafka-client --namespace default -- kafka-console-producer.sh --broker-list kafka-0.kafka-headless.default.svc.cluster.local:9092 --topic sink-topic  

&gt;{&quot;foo&quot;:&quot;bar.....12&quot;}  
&gt;{&quot;foo&quot;:&quot;bar.....1122&quot;}
&gt;

</code></pre>
<p>In mongodb:</p>
<pre><code>rs0:PRIMARY&gt; use mydb;
rs0:PRIMARY&gt; db.sink.find()
{ &quot;_id&quot; : ObjectId(&quot;61e7fb793bb99a00505efa14&quot;), &quot;foo&quot; : &quot;bar.....12&quot; }
{ &quot;_id&quot; : ObjectId(&quot;61e7fb9f3bb99a00505efa16&quot;), &quot;foo&quot; : &quot;bar.....1122&quot; }
rs0:PRIMARY&gt; 

</code></pre>
<p><strong>Issue ?</strong>
The problem is that this chart is working only for mongo-sink. I can not able to use mongo-source-connector to use mongodb as source with above configuration <code>values.yml</code>. <strong>Note</strong>, for using the mongo-source-connector, i do following change in <code>values.yml</code> file.</p>
<pre><code>command: [ &quot;/bin/bash&quot;, &quot;-c&quot;, &quot;/opt/bitnami/kafka/bin/connect-standalone.sh /config/connect-standalone.properties /config/mongodb-source.properties&quot;]
</code></pre>
<p><strong>Expected behavior</strong></p>
<p>(i)- Kafka Connect is working with Mongo as SINK, meaning kafka--&gt;mongo (OK).</p>
<p>(ii) But, it's not working as SOURCE, meaning mongo--&gt;kafka (NOT OK).</p>
<p>There is NO error in pods or any of error messages that tells kafka can not communicate with mongo.</p>
<p><strong>My Questions:</strong></p>
<ol>
<li>If i am missing any necessary settings, what are those to achieve (ii).</li>
<li>Is it possible to run source-connector and sink-connector simulatenously to receieve the updates in kafka and to send the data from kafka. If yes, what the necessary update that are to done to accomplish this.</li>
</ol>
<p>Thanks in advance for your feedback</p>

Faaiz
<blockquote>
<p>The problem is that this chart is working only for mongo-sink</p>
</blockquote>
<p>Because that is the only connector file you gave.</p>
<p>You can give multiple</p>
<pre><code>command: 
  - /bin/bash
  - -c
  - /opt/bitnami/kafka/bin/connect-standalone.sh 
  - /config/connect-standalone.properties 
  - /config/mongodb-source.properties
  - /config/mongodb-sink.properties
</code></pre>
<p>Note: This is not fault-tolernant or to be used in production. Instead, your command should really use <code>connect-distrubuted.sh</code>, and your connector configs can be saved as JSON files and POST-ed to the service ingress port.</p>
