increase notification test reliability by checking for kafka connectors instead of sleep

Issue: ZENKO-5203

Author: SylvainSenechal

tests/ctst/steps/notifications.ts

@@ -4,6 +4,7 @@ import { S3, Utils, AWSVersionObject, NotificationDestination } from 'cli-testin
 import { Consumer, stringDeserializers } from '@platformatic/kafka';
 import Zenko from 'world/Zenko';
 import { putObject } from './utils/utils';
+import { waitForBucketInConnectorPipeline } from './utils/kafka';
 
 const KAFKA_TESTS_TIMEOUT = Number(process.env.KAFKA_TESTS_TIMEOUT) || 60000;
 
@@ -185,8 +186,8 @@ When('i subscribe to {string} notifications for destination {int}',
             this.addCommandParameter({ notificationConfiguration: `'${JSON.stringify(destinationConfig)}'` });
         }
         await S3.putBucketNotificationConfiguration(this.getCommandParameters());
-        // waiting for oplog populator to take the putNotificationConfiguration into account
-        await Utils.sleep(10000);
+        const hosts = this.getSaved<NotificationDestination[]>('notificationDestinations')[destination].hosts;
+        await waitForBucketInConnectorPipeline(hosts, this.getSaved<string>('bucketName'));
     });
 
 When('i subscribe to {string} notifications for destination {int} with {string} filter',
@@ -235,8 +236,8 @@ When('i subscribe to {string} notifications for destination {int} with {string}
             this.addCommandParameter({ notificationConfiguration: `'${JSON.stringify(destinationConfig)}'` });
         }
         await S3.putBucketNotificationConfiguration(this.getCommandParameters());
-        // waiting for oplog populator to take the putNotificationConfiguration into account
-        await Utils.sleep(10000);
+        const hosts = this.getSaved<NotificationDestination[]>('notificationDestinations')[destination].hosts;
+        await waitForBucketInConnectorPipeline(hosts, this.getSaved<string>('bucketName'));
     });
 
 When('i unsubscribe from {string} notifications for destination {int}',

tests/ctst/steps/utils/kafka.ts

@@ -0,0 +1,70 @@
+import { Utils } from 'cli-testing';
+
+function getKafkaConnectUrl(kafkaHosts: string): string {
+    const releasePrefix = kafkaHosts.split('-base-queue-')[0];
+    return `http://${releasePrefix}-base-queue-connector:8083/connectors`;
+}
+
+interface ConnectorInfo {
+    info: {
+        name: string;
+        config: {
+            pipeline?: string;
+            [key: string]: unknown;
+        };
+        [key: string]: unknown;
+    };
+}
+
+
+/**
+ * Polls the Kafka Connect REST API until the given bucket appears in
+ * at least one connector's MongoDB change-stream pipeline (`$match` →
+ * `ns.coll.$in`).  This ensures the oplog-populator has propagated a
+ * new `putBucketNotificationConfiguration` to the connector before the
+ * test proceeds to trigger events.
+ */ 
+export async function waitForBucketInConnectorPipeline(
+    kafkaHosts: string,
+    bucketName: string,
+    timeoutMs = 120000,
+    intervalMs = 1000,
+): Promise<void> {
+    const url = getKafkaConnectUrl(kafkaHosts);
+    const deadline = Date.now() + timeoutMs;
+    let lastConnectorCount = 0;
+    while (Date.now() < deadline) {
+        try {
+            const response = await fetch(`${url}?expand=info`, {
+                signal: AbortSignal.timeout(5000),
+            });
+            const connectors = await response.json() as Record<string, ConnectorInfo>;
+            lastConnectorCount = Object.keys(connectors).length;
+
+            for (const connector of Object.values(connectors)) {
+                const pipelineStr = connector.info?.config?.pipeline;
+                if (!pipelineStr) {
+                    continue;
+                }
+                try {
+                    const pipeline = JSON.parse(pipelineStr) as Array<Record<string, unknown>>;
+                    const matchStage = pipeline[0]?.$match as Record<string, unknown> | undefined;
+                    const nsColl = matchStage?.['ns.coll'] as Record<string, unknown> | undefined;
+                    const bucketList = nsColl?.$in as string[] | undefined;
+                    if (bucketList?.includes(bucketName)) {
+                        return;
+                    }
+                } catch {
+                    // pipeline not valid JSON, skip
+                }
+            }
+        } catch {
+            // Kafka Connect not reachable, retry
+        }
+        await Utils.sleep(intervalMs);
+    }
+    throw new Error(
+        `waitForBucketInConnectorPipeline timed out after ${timeoutMs}ms waiting for bucket ` +
+        `"${bucketName}" in connector pipelines (${lastConnectorCount} connectors checked)`,
+    );
+}