Need help connecting to existing Kafka

[AlexDavydkin]

Hello, I'm trying to configure microcks to interact with existing kafka.

to raise containers with microcks I use the command
docker compose -f docker-compose.yml -f docker-compose-async-addon.yml up -d

I removed the zookiper and kafka raising services from the docker-compose-async-addon.yml file

this is what the file looks like

version: '2'

services:
  app:
    depends_on:
      - mongo
      - keycloak
      - postman
    image: quay.io/microcks/microcks:1.9.0
    container_name: microcks
    volumes:
      - "./config:/deployments/config"
    ports:
      - "8080:8080"
      - "9090:9090"
    environment:
      - SPRING_PROFILES_ACTIVE=prod
      - SPRING_DATA_MONGODB_URI=mongodb://mongo:27017
      - SPRING_DATA_MONGODB_DATABASE=microcks
      - POSTMAN_RUNNER_URL=http://postman:3000
      - TEST_CALLBACK_URL=http://microcks:8080
      - SERVICES_UPDATE_INTERVAL=0 0 0/2 * * *
      - KEYCLOAK_URL=http://keycloak:8080
      - KEYCLOAK_PUBLIC_URL=http://localhost:18080
      - JAVA_OPTIONS=-Dspring.security.oauth2.resourceserver.jwt.issuer-uri=http://localhost:18080/realms/microcks -Dspring.security.oauth2.resourceserver.jwt.jwk-set-uri=http://keycloak:8080/realms/microcks/protocol/openid-connect/certs
      - ASYNC_MINION_URL=http://microcks-async-minion:8081
      - KAFKA_BOOTSTRAP_SERVER=host.docker.internal:9092

  async-minion:
    depends_on:
      - app
    ports:
      - "8081:8081"
    image: quay.io/microcks/microcks-async-minion:1.9.0
    container_name: microcks-async-minion
    restart: on-failure
    volumes:
      - "./config:/deployments/config"
    environment:
      - QUARKUS_PROFILE=docker-compose

Kafka is raised in another container and accessed via url:port localhost:9092

additionally I changed the file install/docker-compose/config/application.properties

# Access to Kafka broker.
%docker-compose.kafka.bootstrap.servers=host.docker.internal:9092
# Do not save any consumer-offset on the broker as there's a re-sync on each minion startup.
%docker-compose.mp.messaging.incoming.microcks-services-updates.enable.auto.commit=false
%docker-compose.mp.messaging.incoming.microcks-services-updates.bootstrap.servers=host.docker.internal:9092
# Explicitly telling the minion the protocols we want to support
%docker-compose.minion.supported-bindings=KAFKA,WS

additionally I changed the file install/docker-compose/config/features.properties

features.feature.async-api.default-binding=KAFKA
features.feature.async-api.endpoint-KAFKA=localhost:9092

and it doesn't work, it gives me an error

2024-04-15 16:02:12 2024-04-15 13:02:12,629 WARN  [org.apa.kaf.cli.NetworkClient] (kafka-producer-network-thread | microcks-async-minion-bytes-producer) [Producer clientId=microcks-async-minion-bytes-producer] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2024-04-15 16:02:13 2024-04-15 13:02:13,412 WARN  [org.apa.kaf.cli.NetworkClient] (smallrye-kafka-consumer-thread-0) [Consumer clientId=kafka-consumer-microcks-services-updates, groupId=microcks-async-minion] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2024-04-15 16:02:13 2024-04-15 13:02:13,413 WARN  [org.apa.kaf.cli.NetworkClient] (kafka-producer-network-thread | microcks-async-minion-str-producer) [Producer clientId=microcks-async-minion-str-producer] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2024-04-15 16:02:13 2024-04-15 13:02:13,731 WARN  [org.apa.kaf.cli.NetworkClient] (kafka-producer-network-thread | microcks-async-minion-bytes-producer) [Producer clientId=microcks-async-minion-bytes-producer] Connection to node 1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
2024-04-15 16:02:14 2024-04-15 13:02:14,384 INFO  [io.git.mic.min.asy.pro.ProducerManager] (QuarkusQuartzScheduler_Worker-18) Producing async mock messages for frequency: 3
2024-04-15 16:02:14 2024-04-15 13:02:14,386 INFO  [io.git.mic.min.asy.pro.KafkaProducerManager] (QuarkusQuartzScheduler_Worker-18) Publishing on topic {UsersignedupAPIv2-0.1.2-TestTopicBroh}, message: {"id": "LXt0lBQs2mou8qKT4ZaNdTXNVeEvmSex", "sendAt": "1713186134386", "fullName": "Laurent Broudoux", "email": "[email protected]", "age": 41} 
2024-04-15 16:02:14 2024-04-15 13:02:14,398 INFO  [io.git.mic.min.asy.pro.KafkaProducerManager] (QuarkusQuartzScheduler_Worker-7) Publishing on topic {UsersignedupAPIv2-0.1.2-TestTopicBroh}, message: {"id":"jYXMcMzwSTQ8vFxmWGgs41ULw0EstTrJ","sendAt":"1713186134398","fullName":"John Doe","email":"[email protected]","age":36} 

at the same time, it is logged as sending to a non-existent topic UsersignedupAPIv2-0.1.2-TestTopicBroh

I use the openapi specification

asyncapi: '2.0.0'
id: 'urn:io.microcks.example.user-signedup-v2'
info:
  title: User signed-up API v2
  version: 0.1.2
  description: Sample AsyncAPI for user signedup events, version 2
defaultContentType: application/json

channels:
  TestTopicBroh:
    description: The topic on which user signed up events may be consumed
    subscribe:
      summary: Receive information about user signed up
      operationId: userSignedUpSubscriptionV2
      message:
        description: An event describing that a user has just signed up.
        traits:
          - $ref: '#/components/messageTraits/commonHeaders'
        payload:
          type: object
          additionalProperties: false
          properties:
            id:
              type: string
            sendAt:
              type: string
            fullName:
              type: string
            email:
              type: string
              format: email
            age:
              type: integer
              minimum: 18
        examples:
          - laurent:
              summary: Example for Laurent user
              headers: |-
                {"my-app-header": 23}
              payload: |-
                {"id": "{{randomString(32)}}", "sendAt": "{{now()}}", "fullName": "Laurent Broudoux", "email": "[email protected]", "age": 41}
          - john:
              summary: Example for John Doe user
              headers:
                my-app-header: 24
              payload:
                id: '{{randomString(32)}}'
                sendAt: '{{now()}}'
                fullName: John Doe
                email: [email protected]
                age: 36

components:
  messageTraits:
    commonHeaders:
      headers:
        type: object
        properties:
          my-app-header:
            type: integer
            minimum: 0
            maximum: 100`

[lbroudoux]

Sorry for the late reply... I just got some time to have a look at it. Will keep you updated ASAP.

[lbroudoux]

So...

The name of the topic UsersignedupAPIv2-0.1.2-TestTopicBroh is expected as it is managed by Microcks for its mocks to avoid collisions with other stuff. The fact that you cannot see it clearly means that Microcks is not able to connect to the Kafka broker.

Now the problem with this setup is certainly due to how the Kafka listener is advertised by the Kafka bootstrap server. From within the Microcks containers, you're using host.docker.internal:9092 that is correct and correspond to the bootstrap server. However, the defautl configuration of bootstrap is to advertise the connection to a Kafka node server that has the same host that the one it is running on - probably localhost:9092. As a consequence, the Microcks containers cannot reach it out because they're translating this to 127.0.0.1 and obviously nothing is running on this port...

So depending on your Kafka implementation, you may be able to configure another listener (on another port) that advertises node server differently and in a way that it can be reached from within the docker network. That's why - most of the time - it's easier to have either Kafka running in the docker network and using different listeners and aliases (that's typically what we're doing) OR to have an external Kafka on another machine that both host and docker networks can resolve the same way.