Introduce liveness-check mechanism for the CDC listener

ballerina/Ballerina.toml

@@ -1,7 +1,7 @@
 [package]
 org = "ballerinax"
 name = "cdc"
-version = "1.1.0"
+version = "1.1.1"
 distribution = "2201.12.0"
 authors = ["Ballerina"]
 repository = "https://github.com/ballerina-platform/module-ballerinax-cdc"
@@ -15,9 +15,9 @@ graalvmCompatible=true
 
 [[platform.java21.dependency]]
 groupId = "io.ballerina.lib.cdc"
-artifactId = "cdc-native-1.1.0"
-version = "1.1.0"
-path = "../native/build/libs/cdc-native-1.1.0.jar"
+artifactId = "cdc-native-1.1.1-SNAPSHOT"
+version = "1.1.1-SNAPSHOT"
+path = "../native/build/libs/cdc-native-1.1.1-SNAPSHOT.jar"
 
 [[platform.java21.dependency]]
 groupId = "io.debezium"

ballerina/CompilerPlugin.toml

@@ -3,4 +3,4 @@ id = "cdc-compiler-plugin"
 class = "io.ballerina.lib.cdc.compiler.CdcCompilerPlugin"
 
 [[dependency]]
-path = "../compiler-plugin/build/libs/cdc-compiler-plugin-1.1.0.jar"
+path = "../compiler-plugin/build/libs/cdc-compiler-plugin-1.1.1-SNAPSHOT.jar"

ballerina/Dependencies.toml

@@ -5,7 +5,7 @@
 
 [ballerina]
 dependencies-toml-version = "2"
-distribution-version = "2201.12.0"
+distribution-version = "2201.13.1"
 
 [[package]]
 org = "ballerina"
@@ -18,7 +18,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "crypto"
-version = "2.9.3"
+version = "2.10.1"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"},
 	{org = "ballerina", name = "time"}
@@ -30,7 +30,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "data.jsondata"
-version = "1.1.0"
+version = "1.1.3"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"},
 	{org = "ballerina", name = "lang.object"}
@@ -144,7 +144,7 @@ dependencies = [
 [[package]]
 org = "ballerina"
 name = "log"
-version = "2.12.0"
+version = "2.16.1"
 dependencies = [
 	{org = "ballerina", name = "io"},
 	{org = "ballerina", name = "jballerina.java"},
@@ -158,7 +158,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "observe"
-version = "1.5.0"
+version = "1.7.0"
 dependencies = [
 	{org = "ballerina", name = "jballerina.java"}
 ]
@@ -179,7 +179,7 @@ modules = [
 [[package]]
 org = "ballerina"
 name = "sql"
-version = "1.16.0"
+version = "1.17.1"
 scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "io"},
@@ -236,7 +236,7 @@ modules = [
 [[package]]
 org = "ballerinax"
 name = "cdc"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
 	{org = "ballerina", name = "crypto"},
 	{org = "ballerina", name = "data.jsondata"},
@@ -277,7 +277,7 @@ dependencies = [
 [[package]]
 org = "ballerinax"
 name = "java.jdbc"
-version = "1.14.0"
+version = "1.14.1"
 scope = "testOnly"
 dependencies = [
 	{org = "ballerina", name = "io"},

ballerina/extern_functions.bal

@@ -41,7 +41,7 @@ public isolated function externDetach(Listener cdcListener, Service cdcService)
 # + cdcListener - the cdc listener object
 # + config - the configuration map containing debezium properties
 # + return - an error if the listener cannot be started, or `()` if successful
-public isolated function externStart(Listener cdcListener, map<string> config) returns Error? = @java:Method {
+public isolated function externStart(Listener cdcListener, map<anydata> config) returns Error? = @java:Method {
     name: "start",
     'class: "io.ballerina.lib.cdc.Listener"
 } external;
@@ -63,3 +63,13 @@ public isolated function externImmediateStop(Listener cdcListener) returns Error
     name: "immediateStop",
     'class: "io.ballerina.lib.cdc.Listener"
 } external;
+
+
+# Checks whether the given CDC listener is live.
+#
+# + cdc - The CDC listener instance to be checked
+# + return - Returns `true` if the listener is considered live, `false` otherwise,
+# or an error if the liveness check fail
+public function isLive(Listener cdc) returns boolean|Error = @java:Method {
+    'class: "io.ballerina.lib.cdc.Listener"
+} external;

ballerina/tests/dynamic_attachment.bal

@@ -15,6 +15,7 @@
 // under the License.
 
 import ballerina/test;
+import ballerina/lang.runtime;
 
 Service testService = service object {
     remote function onCreate(record {} after, string tableName = "") returns error? {
@@ -131,3 +132,87 @@ function testDetachAfterStart() returns error? {
             "Cannot detach a CDC service from the listener once it is running.");
     check mysqlListener.gracefulStop();
 }
+
+@test:Config {
+    groups: ["liveness"]
+}
+function testLivenessBeforeListenerStart() returns error? {
+    MockListener mysqlListener = new ({
+        database: {
+            username,
+            password,
+            port
+        },
+        options: {
+            snapshotMode: NO_DATA
+        }
+    });
+    check mysqlListener.attach(testService);
+    boolean liveness = check isLive(mysqlListener);
+    test:assertFalse(liveness, "Liveness check passes even before listener starts");
+}
+
+@test:Config {
+    groups: ["liveness"]
+}
+function testLivenessWithStartedListener() returns error? {
+    MockListener mysqlListener = new ({
+        database: {
+            username,
+            password,
+            port
+        },
+        options: {
+            snapshotMode: NO_DATA
+        }
+    });
+    check mysqlListener.attach(testService);
+    check mysqlListener.'start();
+    boolean liveness = check isLive(mysqlListener);
+    test:assertTrue(liveness, "Liveness fails for a started listener");
+    check mysqlListener.gracefulStop();
+}
+
+@test:Config {
+    groups: ["liveness"]
+}
+function testLivenessAfterListenerStop() returns error? {
+    MockListener mysqlListener = new ({
+        database: {
+            username,
+            password,
+            port
+        },
+        options: {
+            snapshotMode: NO_DATA
+        }
+    });
+    check mysqlListener.attach(testService);
+    check mysqlListener.'start();
+    check mysqlListener.gracefulStop();
+    boolean liveness = check isLive(mysqlListener);
+    test:assertFalse(liveness, "Liveness check passes after the listener has stopped");
+}
+
+@test:Config {
+    groups: ["liveness"]
+}
+function testLivenessWithoutReceivingEvents() returns error? {
+    MockListener mysqlListener = new ({
+        database: {
+            username,
+            password,
+            port
+        },
+        options: {
+            snapshotMode: NO_DATA
+        },
+        livenessInterval: 5.0
+    });
+    check mysqlListener.attach(testService);
+    check mysqlListener.'start();
+    runtime:sleep(10);
+    boolean liveness = check isLive(mysqlListener);
+    test:assertFalse(liveness, "Liveness check passes even after not receiving events within the liveness interval");
+    check mysqlListener.gracefulStop();
+}

ballerina/tests/mock_listener.bal

@@ -19,21 +19,21 @@ import ballerina/random;
 public isolated class MockListener {
     *Listener;
 
-    private final map<string> & readonly config;
+    private final map<anydata> & readonly config;
     private boolean isStarted = false;
     private boolean hasAttachedService = false;
 
     # Initializes the MySQL listener with the given configuration.
     #
     # + config - The configuration for the MySQL connector
     public isolated function init(*MySqlListenerConfiguration config) {
-        map<string> configMap = {};
+        map<string> debeziumConfigs = {};
         populateDebeziumProperties({
                                        engineName: config.engineName,
                                        offsetStorage: config.offsetStorage,
                                        internalSchemaStorage: config.internalSchemaStorage,
                                        options: config.options
-                                   }, configMap);
+                                   }, debeziumConfigs);
         populateDatabaseConfigurations({
                                            connectorClass: config.database.connectorClass,
                                            hostname: config.database.hostname,
@@ -47,9 +47,13 @@ public isolated class MockListener {
                                            excludedTables: config.database.excludedTables,
                                            includedColumns: config.database.includedColumns,
                                            excludedColumns: config.database.excludedColumns
-                                       }, configMap);
-        configMap["database.server.id"] = "100000";
-        self.config = configMap.cloneReadOnly();
+                                       }, debeziumConfigs);
+        debeziumConfigs["database.server.id"] = "100000";
+        map<anydata> listenerConfigs = {
+            ...debeziumConfigs
+        };
+        listenerConfigs["livenessInterval"] = config.livenessInterval;
+        self.config = listenerConfigs.cloneReadOnly();
     }
 
     # Attaches a CDC service to the MySQL listener.

ballerina/types.bal

@@ -215,9 +215,11 @@ public type Options record {|
 # + internalSchemaStorage - The internal schema history configuration
 # + offsetStorage - The offset storage configuration
 # + options - The additional options for the CDC engine
+# + livenessInterval - Time interval (in seconds) used to evaluate the liveness of the CDC listener
 public type ListenerConfiguration record {|
     string engineName = "ballerina-cdc-connector";
     FileInternalSchemaStorage|KafkaInternalSchemaStorage internalSchemaStorage = {};
     FileOffsetStorage|KafkaOffsetStorage offsetStorage = {};
     Options options = {};
+    decimal livenessInterval = 60.0;
 |};

docs/spec/spec.md

@@ -40,6 +40,7 @@ The conforming implementation of the specification is released and included in t
         - [2.2.4.5 `onTruncate`](#2245-ontruncate)
         - [2.2.4.6 `onError`](#2246-onerror)
       - [2.2.5 Service Configuration](#225-service-configuration)
+    - [2.2 Using health checks](#23-using-health-checks) 
   - [3. Errors](#3-errors)
     - [3.1 Service Error Handling](#31-service-error-handling)
           - [Example: `onCreate` Throwing an Error](#example-oncreate-throwing-an-error)
@@ -289,6 +290,39 @@ remote function onError(cdc:Error e) {
 
 The `cdc:ServiceConfig` annotation can be used to provide additional configurations to the CDC service. These configurations are described in the [Service Configuration](#41-service-config) section.
 
+### 2.3 Using health checks
+
+In a distributed deployment, health checks are essential for monitoring the state of the CDC listener and enabling automated recovery mechanisms. If the listener becomes unhealthy, the orchestrator (e.g., Kubernetes) can restart the service based on the reported health status.
+
+To support this, the `isLive` function is provided to determine the liveness of a given CDC listener instance.
+
+```ballerina
+# Checks whether the given CDC listener is live.
+#
+# + cdc - The CDC listener instance to be checked
+# + return - Returns `true` if the listener is considered live, `false` otherwise,
+#            or an error if the liveness check fails
+public function isLive(cdc: Listener cdc) returns boolean|Error;
+```
+
+The following example demonstrates how the `isLive` function can be integrated with an HTTP-based liveness probe:
+
+```ballerina
+listener cdc:Listener mssqlCdc = ...;
+
+// HTTP-based liveness probe
+service on new http:Listener(...) {
+
+    resource function get liveness() returns http:Ok|http:ServiceUnavailable {
+        boolean isLive = cdc:isLive(mssqlCdc);
+        if isLive {
+            return http:OK;
+        }
+        return http:SERVICE_UNAVAILABLE;
+    }
+}
+```
+
 ## 3. Errors
 
 ### 3.1 Service Error Handling

native/src/main/java/io/ballerina/lib/cdc/BalChangeConsumer.java

@@ -41,10 +41,13 @@
 import io.debezium.engine.ChangeEvent;
 import io.debezium.engine.DebeziumEngine;
 
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static io.ballerina.lib.cdc.utils.Constants.ALLOW_DATA_PROJECTION;
 import static io.ballerina.lib.cdc.utils.Constants.BallerinaErrors.EVENT_PROCESSING_ERROR;
@@ -68,6 +71,7 @@ public class BalChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEv
     private final boolean isSingleServiceAttached;
     private final Service singleService;
     private final Runtime runtime;
+    private final AtomicReference<Instant> lastEventReceivedTime = new AtomicReference<>();
 
     public BalChangeConsumer(Map<String, Service> serviceMap, Runtime runtime) {
         this.serviceMap = new HashMap<>(serviceMap);
@@ -85,6 +89,9 @@ public BalChangeConsumer(Map<String, Service> serviceMap, Runtime runtime) {
     public void handleBatch(List<ChangeEvent<String, String>> records,
                             DebeziumEngine.RecordCommitter<ChangeEvent<String, String>> committer)
             throws InterruptedException {
+        if (!records.isEmpty()) {
+            updateLastEventReceivedTime();
+        }
         for (ChangeEvent<String, String> record : records) {
             Service selectedService = null;
             try {
@@ -221,4 +228,12 @@ private BError createMethodNotFoundError(Payload payload, String methodName) {
         return createError(EVENT_PROCESSING_ERROR, "Function '" + methodName + "' is not available.",
                 null, ValueCreator.createRecordValue(getModule(), EVENT_PROCESSING_ERROR_DETAIL, detail));
     }
+
+    private void updateLastEventReceivedTime() {
+        lastEventReceivedTime.set(Instant.now());
+    }
+
+    public Optional<Instant> getLastEventReceivedTime() {
+        return Optional.ofNullable(lastEventReceivedTime.get());
+    }
 }