add test for debounce by specific fields

.gitignore

@@ -5,4 +5,5 @@ yarn-error.log
 .DS_Store
 .settings.json
 /bazel-*
-.ijwb
+.ijwb
+*.log

src/main/java/configuration/Config.java

@@ -13,6 +13,7 @@ public class Config {
     //Required
     public static String KAFKA_BROKER;
     public static String GROUP_ID;
+    public static String COMMIT_INTERVAL_MS_CONFIG;
     public static String SOURCE_TOPIC;
     public static String TARGET_TOPIC;
 
@@ -43,6 +44,7 @@ public static void init() throws Exception {
 
         KAFKA_BROKER = getString(dotenv, "KAFKA_BROKER");
         GROUP_ID = getString(dotenv, "GROUP_ID");
+        COMMIT_INTERVAL_MS_CONFIG = getOptionalString(dotenv, "COMMIT_INTERVAL_MS_CONFIG", "30000");
 
         // --------------------
         SOURCE_TOPIC = getString(dotenv, "SOURCE_TOPIC");

src/main/java/stream/StreamConfiguration.java

@@ -19,6 +19,7 @@ public class StreamConfiguration extends Properties {
     public StreamConfiguration() {
         super();
         put(StreamsConfig.APPLICATION_ID_CONFIG, Config.GROUP_ID);
+        put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, Config.COMMIT_INTERVAL_MS_CONFIG);
         put(StreamsConfig.CLIENT_ID_CONFIG, Config.GROUP_ID + "_client");
         put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, Config.KAFKA_BROKER);
         put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

tests/jest.config.ts

@@ -21,6 +21,8 @@ const config: JestConfigWithTsJest = {
         // eslint-disable-next-line @typescript-eslint/naming-convention
         '^(\\.{1,2}/.*)\\.js$': '$1',
     },
+    // setupFilesAfterEnv: ['<rootDir>/setupFilesAfterEnv.ts'],
+    testTimeout: 1800000,
 };
 
 export default config;

tests/services/consume.ts

@@ -0,0 +1,17 @@
+import {Kafka, KafkaMessage} from 'kafkajs';
+
+export const consume = async (kafka: Kafka, topic: string, parse = true) => {
+    const consumer = kafka.consumer({groupId: 'orchestrator'});    
+    await consumer.subscribe({topic: topic, fromBeginning: true});
+    const consumedMessage = await new Promise<KafkaMessage>((resolve) => {
+        consumer.run({
+            eachMessage: async ({message}) => resolve(message),
+        });
+    });
+    await consumer.disconnect();
+    const value = parse ? JSON.parse(consumedMessage.value?.toString() ?? '{}') : consumedMessage.value?.toString();
+    const headers = Object.fromEntries(
+        Object.entries(consumedMessage.headers!).map(([key, value]) => [key, value?.toString()])
+    );
+    return {value, headers};
+};

tests/services/getOffset.ts

@@ -0,0 +1,9 @@
+import {Kafka} from 'kafkajs';
+
+export const getOffset = async (kafka: Kafka, groupId: string, topic: string) => {
+    const admin = kafka.admin();
+    await admin.connect();
+    const metadata = await admin.fetchOffsets({groupId, topics: [topic]});
+    admin.disconnect();
+    return Number.parseInt(metadata[0]?.partitions[0]?.offset!);
+};

tests/services/produce.ts

@@ -0,0 +1,8 @@
+import {ProducerRecord} from 'kafkajs';
+import {Orchestrator} from '../testcontainers/orchestrator.js';
+
+export const produce = async (orchestrator: Orchestrator, record: ProducerRecord) => {
+    const producer = orchestrator.kafkaClient.producer();
+    await producer.connect();
+    await producer.send(record);
+};

tests/setupFilesAfterEnv.ts

@@ -0,0 +1,3 @@
+import {jest} from '@jest/globals';
+
+jest.retryTimes(3, {logErrorsBeforeRetry: true});

tests/specs/__snapshots__/debounceByKey.spec.ts.snap

@@ -0,0 +1,10 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`tests should debounce by key 1`] = `
+{
+  "headers": {},
+  "value": {
+    "data": "bar",
+  },
+}
+`;

tests/specs/__snapshots__/debounceBySpecificFields.spec.ts.snap

@@ -0,0 +1,12 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`tests should debounce by specific fields 1`] = `
+{
+  "headers": {},
+  "value": {
+    "data": {
+      "type": "bar",
+    },
+  },
+}
+`;

tests/specs/debounceByKey.spec.ts

@@ -0,0 +1,50 @@
+import delay from 'delay';
+
+import type {Orchestrator} from '../testcontainers/orchestrator.js';
+import {start} from '../testcontainers/orchestrator.js';
+import {produce} from '../services/produce.js';
+import {getOffset} from '../services/getOffset.js';
+import {consume} from '../services/consume.js';
+
+const groupId = 'the-consumer';
+const sourceTopic = 'test-input';
+const targetTopic = 'test-output';
+
+describe('tests', () => {
+    let orchestrator: Orchestrator;
+
+    beforeEach(
+        async () => {
+            orchestrator = await start(
+                {
+                    KAFKA_BROKER: 'kafka:9092',
+                    MONITORING_SERVER_PORT: '3000',
+                    GROUP_ID: groupId,
+                    COMMIT_INTERVAL_MS_CONFIG: '1000',
+                    SOURCE_TOPIC: sourceTopic,
+                    TARGET_TOPIC: targetTopic,
+                    WINDOW_DURATION: '5',
+                },
+                [sourceTopic, targetTopic]
+            );
+        },
+        5 * 60 * 1000
+    );
+
+    afterEach(async () => {
+        if (!orchestrator) {
+            return;
+        }
+        await orchestrator.stop();
+    });
+
+    it('should debounce by key', async () => {
+        await produce(orchestrator, {topic: sourceTopic, messages: [{value: JSON.stringify({data: 'bar'}), key: 'barkey'}]});
+        await produce(orchestrator, {topic: sourceTopic, messages: [{value: JSON.stringify({data: 'bar'}), key: 'barkey'}]});
+
+        await delay(5000);
+
+        await expect(consume(orchestrator.kafkaClient, targetTopic)).resolves.toMatchSnapshot();
+        await expect(getOffset(orchestrator.kafkaClient, groupId, sourceTopic)).resolves.toBe(2);
+    });
+});

tests/specs/debounceBySpecificFields.spec.ts

@@ -0,0 +1,52 @@
+import delay from 'delay';
+
+import type {Orchestrator} from '../testcontainers/orchestrator.js';
+import {start} from '../testcontainers/orchestrator.js';
+import {produce} from '../services/produce.js';
+import {getOffset} from '../services/getOffset.js';
+import {consume} from '../services/consume.js';
+
+const groupId = 'test11';
+const sourceTopic = 'test-input';
+const targetTopic = 'test-output';
+
+describe('tests', () => {
+    let orchestrator: Orchestrator;
+
+    beforeEach(
+        async () => {
+            orchestrator = await start(
+                {
+                    KAFKA_BROKER: 'kafka:9092',
+                    MONITORING_SERVER_PORT: '3000',
+                    GROUP_ID: groupId,
+                    COMMIT_INTERVAL_MS_CONFIG: '1000',
+                    SOURCE_TOPIC: sourceTopic,
+                    TARGET_TOPIC: targetTopic,
+                    WINDOW_DURATION: '5',
+                    OUTPUT_PARTITIONING_KEY: 'SPECIFIC_FIELDS',
+                    OUTPUT_PARTITIONING_KEY_FIELDS: '/data/type',
+                },
+                [sourceTopic, targetTopic]
+            );
+        },
+        5 * 60 * 1000
+    );
+
+    afterEach(async () => {
+        if (!orchestrator) {
+            return;
+        }
+        await orchestrator.stop();
+    });
+
+    it('should debounce by specific fields', async () => {
+        await produce(orchestrator, {topic: sourceTopic, messages: [{value: JSON.stringify({data: {type: 'bar'}}), key: '1'}]});
+        await produce(orchestrator, {topic: sourceTopic, messages: [{value: JSON.stringify({data: {type: 'bar'}}), key: '2'}]});
+
+        await delay(5000);
+
+        await expect(consume(orchestrator.kafkaClient, targetTopic)).resolves.toMatchSnapshot();
+        await expect(getOffset(orchestrator.kafkaClient, groupId, sourceTopic)).resolves.toBe(2);
+    });
+});

tests/testcontainers/dafka.ts

@@ -0,0 +1,35 @@
+import {StartedNetwork, StoppedTestContainer, Wait} from 'testcontainers';
+import {GenericContainer} from 'testcontainers';
+import fs from 'node:fs';
+
+const startupTimeout = parseInt(process.env.STARTUP_TIMEOUT ?? '60000');
+
+export interface ServiceContainer {
+    stop: () => Promise<StoppedTestContainer>;
+}
+
+export const dafka = async (
+    network: StartedNetwork,
+    env: Record<string, string>
+): Promise<ServiceContainer> => {
+    const container = await new GenericContainer('bazel/src:image')
+        .withExposedPorts(3000)
+        .withNetwork(network)
+        .withEnvironment(env)
+        .withWaitStrategy(Wait.forLogMessage(`dafka-debounce-${env.GROUP_ID} started`))
+        .withStartupTimeout(startupTimeout)
+        .start();
+
+    if (process.env.DEBUG) {
+        try {
+            fs.truncateSync('service.log', 0);
+        } catch (err) {
+            fs.writeFileSync('service.log', "", { flag: "wx" });
+        }
+        await container.logs().then((logs) => logs.pipe(fs.createWriteStream('service.log')));
+    }
+
+    return {
+        stop: () => container.stop(),
+    };
+};