Introduce On error handler

ballerina/error.bal

@@ -37,6 +37,24 @@ public type InvalidConfigError distinct Error;
 # temporary file locks (450), or server-side processing errors (451).
 public type ServiceUnavailableError distinct Error;
 
+# Represents an error that occurs when file content cannot be converted to the expected type.
+# This includes JSON/XML parsing errors, CSV format errors, and record type binding failures.
+# This error type is applicable to both Client operations and Listener callbacks.
+#
+# When used with the Listener, if an `onError` remote function is defined in the service,
+# it will be invoked with this error type, allowing for custom error handling such as
+# moving failed files to an error folder or sending notifications.
+public type ContentBindingError distinct Error & error<ContentBindingErrorDetail>;
+
+# Detail record for ContentBindingError providing additional context about the binding failure.
+#
+# + filePath - The file path that caused the error
+# + content - The raw file content as bytes that failed to bind
+public type ContentBindingErrorDetail record {|
+    string filePath?;
+    byte[] content?;
+|};
+
 # Represents an error that occurs when all retry attempts have been exhausted.
 # This error wraps the last failure encountered during retry attempts.
 public type AllRetryAttemptsFailedError distinct Error;

ballerina/tests/client_endpoint_test.bal

@@ -323,6 +323,85 @@ function testPutXmlFailure() returns error? {
     test:assertTrue(got is Error, msg = "XML content binding should have failed for non-XML content");
 }
 
+// Test that getJson returns ContentBindingError when JSON parsing fails
+@test:Config {dependsOn: [testPutFileContent]}
+function testGetJsonContentBindingError() returns error? {
+    string path = "/home/in/invalid-json.txt";
+    // Write invalid JSON content
+    check (<Client>clientEp)->putText(path, "this is not valid json {{{");
+
+    json|Error result = (<Client>clientEp)->getJson(path);
+    test:assertTrue(result is ContentBindingError, msg = "getJson should return ContentBindingError for invalid JSON");
+    if result is ContentBindingError {
+        test:assertTrue(result.detail().filePath is string, msg = "ContentBindingError should contain filePath");
+        test:assertTrue(result.detail().content is byte[], msg = "ContentBindingError should contain content bytes");
+    }
+
+    // Cleanup
+    check (<Client>clientEp)->delete(path);
+}
+
+// Test that getXml returns ContentBindingError when XML parsing fails
+@test:Config {dependsOn: [testPutFileContent]}
+function testGetXmlContentBindingError() returns error? {
+    string path = "/home/in/invalid-xml.txt";
+    // Write invalid XML content
+    check (<Client>clientEp)->putText(path, "this is not valid xml <<<>>>");
+
+    xml|Error result = (<Client>clientEp)->getXml(path);
+    test:assertTrue(result is ContentBindingError, msg = "getXml should return ContentBindingError for invalid XML");
+    if result is ContentBindingError {
+        test:assertTrue(result.detail().filePath is string, msg = "ContentBindingError should contain filePath");
+        test:assertTrue(result.detail().content is byte[], msg = "ContentBindingError should contain content bytes");
+    }
+
+    // Cleanup
+    check (<Client>clientEp)->delete(path);
+}
+
+// Test that getJson with typed binding returns ContentBindingError when type doesn't match
+@test:Config {dependsOn: [testPutFileContent]}
+function testGetJsonTypedContentBindingError() returns error? {
+    string path = "/home/in/json-typed-binding-error.json";
+    // Write JSON missing required field 'age'
+    json content = {name: "Alice"};
+    check (<Client>clientEp)->putJson(path, content);
+
+    // Try to bind to a record that requires 'age' field
+    PersonStrict|Error result = (<Client>clientEp)->getJson(path);
+    test:assertTrue(result is ContentBindingError, msg = "getJson should return ContentBindingError when type binding fails");
+    if result is ContentBindingError {
+        test:assertTrue(result.detail().filePath is string, msg = "ContentBindingError should contain filePath");
+        test:assertTrue(result.detail().content is byte[], msg = "ContentBindingError should contain content bytes");
+    }
+
+    // Cleanup
+    check (<Client>clientEp)->delete(path);
+}
+
+// Test that getCsv returns ContentBindingError when type binding fails
+@test:Config {dependsOn: [testPutFileContent]}
+function testGetCsvContentBindingError() returns error? {
+    string path = "/home/in/csv-binding-error.csv";
+    // Write CSV with non-integer value in age column
+    string[][] csvData = [
+        ["name", "age"],
+        ["Alice", "not-a-number"]  // Invalid: age should be int
+    ];
+    check (<Client>clientEp)->putCsv(path, csvData);
+
+    // Try to bind to a record that expects 'age' as int
+    CsvPersonStrict[]|Error result = (<Client>clientEp)->getCsv(path);
+    test:assertTrue(result is ContentBindingError, msg = "getCsv should return ContentBindingError when type binding fails");
+    if result is ContentBindingError {
+        test:assertTrue(result.detail().filePath is string, msg = "ContentBindingError should contain filePath");
+        test:assertTrue(result.detail().content is byte[], msg = "ContentBindingError should contain content bytes");
+    }
+
+    // Cleanup
+    check (<Client>clientEp)->delete(path);
+}
+
 @test:Config {dependsOn: [testPutFileContent]}
 function testPutText() returns error? {
     string txt = "hello text content";
@@ -1133,14 +1212,15 @@ public function testListFiles() {
         "childDirectory",
         "delete",
         "test2.txt",
+        "onerror-tests",
         "test4.txt",
         "child_directory",
         "content-methods",
         "age",
         "retry",
         "test3.txt"
     ];
-    int[] fileSizes = [0, 61, 0, 0, 0, 0, 0, 145, 0, 0, 16400, 9000, 0, 0, 0, 0, 12];
+    int[] fileSizes = [0, 61, 0, 0, 0, 0, 0, 145, 0, 0, 16400, 0, 9000, 0, 0, 0, 0, 12];
     FileInfo[]|Error response = (<Client>clientEp)->list("/home/in");
     if response is FileInfo[] {
         log:printInfo("List of files/directories: ");

ballerina/tests/listener_on_error_test.bal

@@ -0,0 +1,240 @@
+// Copyright (c) 2026 WSO2 LLC. (http://www.wso2.com) All Rights Reserved.
+//
+// WSO2 LLC. licenses this file to you under the Apache License,
+// Version 2.0 (the "License"); you may not use this file except
+// in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import ballerina/lang.runtime;
+import ballerina/log;
+import ballerina/test;
+
+// Record type for testing JSON binding errors
+type StrictPerson record {|
+    string name;
+    int age;
+    string email;  // Required field that's missing in test JSON
+|};
+
+// Global tracking variables for onError tests
+boolean onErrorInvoked = false;
+ContentBindingError? lastBindingError = ();
+string lastErrorFilePath = "";
+int onErrorInvocationCount = 0;
+
+// Directory for onError tests
+const string ON_ERROR_TEST_DIR = "/home/in/onerror-tests";
+
+@test:Config {
+    enable: true
+}
+public function testOnErrorBasic() returns error? {
+    // Reset state
+    onErrorInvoked = false;
+    lastBindingError = ();
+    lastErrorFilePath = "";
+    onErrorInvocationCount = 0;
+
+    // Service with onFileJson that will fail and onError handler
+    Service onErrorService = service object {
+        remote function onFileJson(StrictPerson content, FileInfo fileInfo, Caller caller) returns error? {
+            // This should not be called since binding will fail
+            log:printInfo(string `onFileJson invoked for: ${fileInfo.name}`);
+        }
+
+        remote function onError(Error err, Caller caller) returns error? {
+            log:printInfo("onError invoked");
+            log:printError("Binding error", err);
+            onErrorInvoked = true;
+            // Verify that the error is a ContentBindingError
+            if err is ContentBindingError {
+                lastBindingError = err;
+                // Access filePath from error detail
+                lastErrorFilePath = err.detail().filePath ?: "";
+            }
+            onErrorInvocationCount += 1;
+        }
+    };
+
+    // Create listener
+    Listener onErrorListener = check new ({
+        protocol: FTP,
+        host: "127.0.0.1",
+        auth: {credentials: {username: "wso2", password: "wso2123"}},
+        port: 21212,
+        path: ON_ERROR_TEST_DIR,
+        pollingInterval: 4,
+        fileNamePattern: "errortest.*\\.json"
+    });
+
+    check onErrorListener.attach(onErrorService);
+    check onErrorListener.'start();
+    runtime:registerListener(onErrorListener);
+
+    // Upload invalid JSON file (missing required field 'email')
+    check (<Client>clientEp)->putText(ON_ERROR_TEST_DIR + "/errortest.json",
+        "{\"name\": \"John\", \"age\": 30}");
+    runtime:sleep(10);
+
+    // Cleanup
+    runtime:deregisterListener(onErrorListener);
+    check onErrorListener.gracefulStop();
+
+    test:assertTrue(onErrorInvoked, "onError should have been invoked");
+    test:assertTrue(lastBindingError is ContentBindingError, "Should have received ContentBindingError");
+    test:assertTrue(lastErrorFilePath.endsWith(".json"), "Error file path should end with .json");
+}
+
+@test:Config {
+    enable: true,
+    dependsOn: [testOnErrorBasic]
+}
+public function testOnErrorWithMinimalParameters() returns error? {
+    // Reset state
+    onErrorInvoked = false;
+    lastBindingError = ();
+    onErrorInvocationCount = 0;
+
+    // Service with onError handler with only error parameter
+    Service minimalOnErrorService = service object {
+        remote function onFileJson(StrictPerson content, FileInfo fileInfo) returns error? {
+            // This should not be called since binding will fail
+            log:printInfo(string `onFileJson invoked for: ${fileInfo.name}`);
+        }
+
+        remote function onError(Error err) returns error? {
+            log:printInfo("onError (minimal) invoked");
+            log:printError("Binding error", err);
+            onErrorInvoked = true;
+            // Verify that the error is a ContentBindingError
+            if err is ContentBindingError {
+                lastBindingError = err;
+            }
+            onErrorInvocationCount += 1;
+        }
+    };
+
+    // Create listener
+    Listener minimalListener = check new ({
+        protocol: FTP,
+        host: "127.0.0.1",
+        auth: {credentials: {username: "wso2", password: "wso2123"}},
+        port: 21212,
+        path: ON_ERROR_TEST_DIR,
+        pollingInterval: 4,
+        fileNamePattern: "minimal.*\\.json"
+    });
+
+    check minimalListener.attach(minimalOnErrorService);
+    check minimalListener.'start();
+    runtime:registerListener(minimalListener);
+
+    // Upload invalid JSON
+    check (<Client>clientEp)->putText(ON_ERROR_TEST_DIR + "/minimal.json",
+        "{\"name\": \"Jane\", \"age\": 25}");
+    runtime:sleep(10);
+
+    // Cleanup
+    runtime:deregisterListener(minimalListener);
+    check minimalListener.gracefulStop();
+
+    test:assertTrue(onErrorInvoked, "onError (minimal) should have been invoked");
+    test:assertTrue(lastBindingError is ContentBindingError, "Should have received ContentBindingError");
+}
+
+@test:Config {
+    enable: true,
+    dependsOn: [testOnErrorWithMinimalParameters]
+}
+public function testOnErrorWithCallerOperations() returns error? {
+    // Reset state
+    onErrorInvoked = false;
+    lastBindingError = ();
+    lastErrorFilePath = "";
+    onErrorInvocationCount = 0;
+
+    // Service that uses caller to perform operations on failed files
+    Service moveOnErrorService = service object {
+        remote function onFileJson(StrictPerson content, FileInfo fileInfo, Caller caller) returns error? {
+            log:printInfo(string `onFileJson invoked for: ${fileInfo.name}`);
+        }
+
+        remote function onError(Error err, Caller caller) returns error? {
+            onErrorInvoked = true;
+            onErrorInvocationCount += 1;
+
+            // Verify that the error is a ContentBindingError and access details
+            if err is ContentBindingError {
+                string? filePath = err.detail().filePath;
+                log:printInfo(string `onError invoked for file: ${filePath ?: "unknown"}`);
+                lastBindingError = err;
+                lastErrorFilePath = filePath ?: "";
+
+                // Try to delete the problematic file using caller
+                if filePath is string {
+                    Error? deleteResult = caller->delete(filePath);
+                    if deleteResult is () {
+                        log:printInfo("Successfully deleted error file");
+                    } else {
+                        log:printError("Failed to delete error file", deleteResult);
+                    }
+                }
+            } else {
+                log:printError("Unexpected error type", err);
+            }
+        }
+    };
+
+    // Create listener
+    Listener moveListener = check new ({
+        protocol: FTP,
+        host: "127.0.0.1",
+        auth: {credentials: {username: "wso2", password: "wso2123"}},
+        port: 21212,
+        path: ON_ERROR_TEST_DIR,
+        pollingInterval: 4,
+        fileNamePattern: "moveerror.*\\.json"
+    });
+
+    check moveListener.attach(moveOnErrorService);
+    check moveListener.'start();
+    runtime:registerListener(moveListener);
+
+    // Upload invalid JSON
+    check (<Client>clientEp)->putText(ON_ERROR_TEST_DIR + "/moveerror.json",
+        "{\"name\": \"Bob\", \"age\": 40}");
+    runtime:sleep(10);
+
+    // Cleanup
+    runtime:deregisterListener(moveListener);
+    check moveListener.gracefulStop();
+
+    test:assertTrue(onErrorInvoked, "onError should have been invoked");
+    test:assertTrue(lastBindingError is ContentBindingError, "Should have received ContentBindingError");
+}
+
+@test:Config {
+    enable: true,
+    dependsOn: [testOnErrorWithCallerOperations]
+}
+public function testContentBindingErrorType() returns error? {
+    // Test that ContentBindingError is a distinct error type with detail record
+    ContentBindingError testError = error ContentBindingError("Test binding error",
+        filePath = "/test/file.json",
+        content = [72, 101, 108, 108, 111]  // "Hello" as bytes
+    );
+
+    test:assertTrue(testError is Error, "ContentBindingError should be a subtype of Error");
+    test:assertEquals(testError.message(), "Test binding error", "Error message should match");
+    test:assertEquals(testError.detail().filePath, "/test/file.json", "filePath should match");
+    test:assertEquals(testError.detail().content, [72, 101, 108, 108, 111], "content should match");
+}

changelog.md

@@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ## unreleased
 
 ### Added
+- [Introduce onError remote function](https://github.com/ballerina-platform/ballerina-library/issues/8605)
 - [Add automatic retry support with exponential backoff for FTP client](https://github.com/ballerina-platform/ballerina-library/issues/8585)
 - [Add FTP Listener Coordination Support](https://github.com/ballerina-platform/ballerina-library/issues/8490)
 - [Add distinct error types](https://github.com/ballerina-platform/ballerina-library/issues/8597)