Debezium Server with NATS JetStream No Responders error

I tried to setup CDC with NATS JetStream. I get the following error:

Error Publishing: 503 No Responders Available For Request

Tell me where did I go wrong?

Debezium full log: https://logpaste.com/6IRk5ixw

NATS log: https://logpaste.com/Yr6OTzqm

To reproduce:

1. Install go-task
2. Install docker-compose
3. Copy application.properties to conf/debezium/application.properties
4. Copy inst.sql to mssql/init.sql
5. Exec the following commands.

task init
task debezium
# tack clean # to clean up

My setup and configs.

docker-compose.yaml

services:
  nats:
    image: nats:${NATS_VERSION}
    ports:
      - 4222:4222
      - 8222:8222
    command:
      - "--debug"
      - "--http_port=8222"
      - "--jetstream"
    healthcheck:
      test: wget --no-verbose --tries=1 --spider http://localhost:8222/healthz || exit 1
      interval: 10s
      timeout: 5s
      retries: 3

  mssql:
    image: mcr.microsoft.com/mssql/server:${MSSQL_VERSION}
    hostname: mssql
    ports:
      - 1433:1433
    environment:
      ACCEPT_EULA: Y
      MSSQL_PID: Developer
      MSSQL_AGENT_ENABLED: True
      MSSQL_COLLATION: Cyrillic_General_CI_AS
      MSSQL_SA_PASSWORD: ${MSSQL_SA_PASSWORD}
    healthcheck:
      test: /opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "${MSSQL_SA_PASSWORD}" -Q "SELECT 1" || exit 1
      interval: 10s
      timeout: 5s
      retries: 3

  debezium:
    image: debezium/server:${DEBEZIUM_VERSION}
    ports:
      - 8080:8080
    depends_on:
      - nats
      - mssql
    volumes:
      - $PWD/conf/debezium/application.properties:/debezium/conf/application.properties
      # - $PWD/data:/data/offsets.dat
    healthcheck:
      test: curl --silent http://localhost:8080/q/health || exit 1
      interval: 10s
      timeout: 5s
      retries: 3

application.properties:

debezium.source.connector.class=io.debezium.connector.sqlserver.SqlServerConnector
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.schema.history.internal=io.debezium.storage.file.history.FileSchemaHistory
debezium.source.schema.history.internal.file.filename=data/schema_history.dat
debezium.source.topic.prefix=mssql
debezium.source.database.hostname=mssql
debezium.source.database.port=1433
debezium.source.database.user=dbz
debezium.source.database.password=pass
debezium.source.database.names=test_db
debezium.source.database.applicatiotnIntent=ReadOnly
debezium.source.database.encrypt=false
debezium.source.table.include.list=dbo.test_1,dbo.test_2
debezium.sink.type=nats-jetstream
debezium.sink.nats-jetstream.url=nats://nats:4222
debezium.sink.nats-jetstream.create-stream=true
debezium.sink.nats-jetstream.subjects=*.*.*
debezium.sink.nats-jetstream.storage=file
quarkus.log.level=DEBUG
quarkus.log.console.json=false

init.sql:

CREATE DATABASE [test_db];

ALTER DATABASE [test_db] SET ALLOW_SNAPSHOT_ISOLATION ON;

GO

USE [test_db];

CREATE LOGIN dbz WITH PASSWORD = 'pass', CHECK_POLICY = OFF;
CREATE USER dbz FOR LOGIN dbz;

EXEC sys.sp_cdc_enable_db;

GRANT SELECT ON SCHEMA :: dbo TO dbz;
GRANT SELECT ON SCHEMA :: cdc TO dbz;

-- DROP TABLE [test_1];

CREATE TABLE [test_1] (
    id int,
    comment text,
    PRIMARY KEY(id)
);

GRANT SELECT ON dbo.[test_1] TO dbz;

EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'test_1', @role_name = NULL, @supports_net_changes = 0;

INSERT INTO [test_1] (id, comment) VALUES (1, 'text'), (2, 'text'), (3, 'text'), (4, 'text'), (5, 'text');

GO

Taskfile.yml:

version: "3"

env:
  NATS_VERSION: "2.10-alpine"
  DEBEZIUM_VERSION: "2.6.2.Final"
  MSSQL_VERSION: "2019-latest"
  POSTGRES_VERSION: "16.3"
  POSTGRES_USER: user
  POSTGRES_PASSWORD: Password*
  POSTGRES_DB: test_db
  MSSQL_SA_PASSWORD: Password*
  MSSQL_DB: test_db

tasks:
  up:
    run: once
    cmds:
      - docker-compose up --wait --detach

  down:
    run: once
    cmds:
      - docker-compose stop

  clean:
    deps:
      - down
    cmds:
      - docker-compose rm -f

  init:
    run: once
    deps:
      - task: nats:init
      - task: mssql:init
      # - docker-compose up --wait --detach debezium

  mssql:up:
    cmds:
      - docker-compose up --wait --detach mssql

  mssql:init:
    deps:
      - task: mssql:up
    env:
      SQLCMDUSER: sa
      SQLCMDPASSWORD: "{{.MSSQL_SA_PASSWORD}}"
      SQLCMDSERVER: localhost
    cmds:
      - sqlcmd --input-file mssql/init.sql

  nats:up:
    cmds:
      - docker-compose up --wait --detach nats

  nats:init:
    deps:
      - task: nats:up
    cmds:
      - nats -s nats://127.0.0.1:4222 stream add --config conf/nats/stream.json

  debezium:
    cmds:
      - docker-compose up debezium --no-log-prefix

I found solution. The reason was an attempt to write to a non-existent topic.

We should define topic.prefix as subject and all other subjects should comply with the following scheme: <topic.prefix>.<database>.<schema>.<table>.

debezium.source.topic.prefix=src
debezium.source.database.names=test_db
debezium.sink.nats-jetstream.create-stream=true
debezium.sink.nats-jetstream.subjects=src,src.*.*.*
# or
debezium.sink.nats-jetstream.subjects=src,src.test_db.*.*
# or
debezium.sink.nats-jetstream.subjects=src,src.test_db.dbo.*

Also we can disable stream creation and use config:

{
  "config": {
    "name": "dbz-mssql",
    "subjects": ["src", "src.*.*.*"],
    "retention": "limits",
    "max_consumers": -1,
    "max_msgs": -1,
    "max_bytes": -1,
    "max_age": 31536000000000000,
    "max_msg_size": -1,
    "storage": "file",
    "discard": "old",
    "num_replicas": 1,
    "duplicate_window": 120000000000,
    "compression": "s2"
  }
}

nats stream add --config config.json