test: add integration test for karapace (#15486)

ci/scripts/gen-integration-test-yaml.py

@@ -34,7 +34,7 @@
     'big-query-sink': ['json'],
     'mindsdb': ['json'],
     'vector': ['json'],
-    'nats': ['json'],
+    'nats': ['json', 'protobuf'],
     'doris-sink': ['json'],
     'starrocks-sink': ['json'],
     'deltalake-sink': ['json'],

ci/scripts/notify.py

@@ -87,6 +87,8 @@
     "deltalake-sink-json": ["xinhao"],
     "pinot-sink-json": ["yiming"],
     "client-library-none": ["tao"],
+    "nats-json": ["tao"],
+    "nats-protobuf": ["tao"],
 }
 
 def get_failed_tests(get_test_status, test_map):

integration_tests/nats/README.md

@@ -37,10 +37,23 @@ psql -h localhost -p 4566 -d dev -U root
 
 
 3. test source:
-```sh
-// connect to risingwave and manually execute the sql file create_source.sql
-psql -h localhost -p 4566 -d dev -U root
 
+Connect to risingwave and manually execute the following SQL.
+
+```sql
+CREATE TABLE
+  nats_source_table (id integer, name varchar)
+WITH
+  (
+    connector = 'nats',
+    server_url = 'nats-server:4222',
+    subject = 'subject1',
+    stream = 'my_stream',
+    connect_mode = 'plain'
+  ) FORMAT PLAIN ENCODE JSON;
+```
+
+```sh
 // publish data into nats
 nats pub subject2 --server=localhost:4222 --count=20 "{\"id\":{{Count}},\"name\":\"Alice{{Count}}\"}"
 

integration_tests/nats/create_mv.sql

@@ -0,0 +1,32 @@
+CREATE TABLE
+  personnel (id integer, name varchar);
+
+
+CREATE SINK nats_sink
+FROM
+  personnel
+WITH
+  (
+    connector = 'nats',
+    server_url = 'nats-server:4222',
+    subject = 'subject1',
+    type = 'append-only',
+    force_append_only = 'true',
+    connect_mode = 'plain'
+  );
+
+
+INSERT INTO
+  personnel
+VALUES
+  (1, 'Alice'),
+  (2, 'Bob'),
+  (3, 'Tom'),
+  (4, 'Jerry'),
+  (5, 'Araminta'),
+  (6, 'Clover'),
+  (7, 'Posey'),
+  (8, 'Waverly');
+
+
+FLUSH;

integration_tests/nats/create_source.sql

@@ -1,49 +1,3 @@
-CREATE TABLE
-  personnel (id integer, name varchar);
-
-
-CREATE TABLE
-  nats_source_table (id integer, name varchar)
-WITH
-  (
-    connector = 'nats',
-    server_url = 'nats-server:4222',
-    subject = 'subject1',
-    stream = 'my_stream',
-    connect_mode = 'plain'
-  ) FORMAT PLAIN ENCODE JSON;
-
-
-CREATE SINK nats_sink
-FROM
-  personnel
-WITH
-  (
-    connector = 'nats',
-    server_url = 'nats-server:4222',
-    subject = 'subject1',
-    type = 'append-only',
-    force_append_only = 'true',
-    connect_mode = 'plain'
-  );
-
-
-INSERT INTO
-  personnel
-VALUES
-  (1, 'Alice'),
-  (2, 'Bob'),
-  (3, 'Tom'),
-  (4, 'Jerry'),
-  (5, 'Araminta'),
-  (6, 'Clover'),
-  (7, 'Posey'),
-  (8, 'Waverly');
-
-
-FLUSH;
-
-
 CREATE TABLE live_stream_metrics (
   client_ip VARCHAR,
   user_agent VARCHAR,

integration_tests/nats/data_check

@@ -1 +1 @@
-personnel,nats_source_table,live_stream_metrics
+personnel,live_stream_metrics

integration_tests/nats/pb/create_source.sql

@@ -9,4 +9,4 @@ WITH
   ) FORMAT PLAIN ENCODE PROTOBUF (
     message = 'livestream.schema.LiveStreamMetrics',
     schema.location = 'http://file_server:8080/schema'
-  );
+  );

integration_tests/nats/query.sql

@@ -1,8 +1,6 @@
-select
+SELECT
     *
-from
-    nats_source_table
-order by
-    id
+FROM
+    live_stream_metrics
 LIMIT
     10;

integration_tests/schema-registry/create_source.sql

@@ -4,4 +4,12 @@ CREATE SOURCE student WITH (
     properties.bootstrap.server = 'message_queue:29092',
     scan.startup.mode = 'earliest'
 )
-FORMAT PLAIN ENCODE AVRO (schema.registry = 'http://message_queue:8081');
+FORMAT PLAIN ENCODE AVRO (schema.registry = 'http://message_queue:8081');
+
+CREATE SOURCE student_karapace WITH (
+    connector = 'kafka',
+    topic = 'sr-test',
+    properties.bootstrap.server = 'message_queue:29092',
+    scan.startup.mode = 'earliest'
+)
+FORMAT PLAIN ENCODE AVRO (schema.registry = 'http://karapace-registry:8081');

integration_tests/schema-registry/data_check

@@ -1 +1 @@
-student_view
+student_view,student_karapace

integration_tests/schema-registry/docker-compose.yml

@@ -27,7 +27,7 @@ services:
       service: message_queue
   datagen:
     image: python:3.10
-    depends_on: [ message_queue ]
+    depends_on: [message_queue]
     volumes:
       - type: bind
         source: ./datagen.py
@@ -38,8 +38,27 @@ services:
       - |
         pip install requests fastavro confluent_kafka
         python /datagen.py message_queue:29092 http://message_queue:8081 sr-test
+        python /datagen.py message_queue:29092 http://karapace-registry:8081 sr-test
     restart: always
     container_name: datagen
+  karapace-registry:
+    image: ghcr.io/aiven-open/karapace:latest
+    entrypoint:
+      - /bin/bash
+      - /opt/karapace/start.sh
+      - registry
+    environment:
+      KARAPACE_ADVERTISED_HOSTNAME: karapace-registry
+      KARAPACE_BOOTSTRAP_URI: message_queue:29092
+      KARAPACE_PORT: 8081
+      KARAPACE_HOST: 0.0.0.0
+      KARAPACE_CLIENT_ID: karapace
+      KARAPACE_GROUP_ID: karapace-registry
+      KARAPACE_MASTER_ELIGIBILITY: "true"
+      KARAPACE_TOPIC_NAME: _schemas
+      KARAPACE_LOG_LEVEL: WARNING
+      KARAPACE_COMPATIBILITY: FULL
+    container_name: karapace-registry
 volumes:
   risingwave-standalone:
     external: false