diff --git a/ingestion/setup.py b/ingestion/setup.py
index c243d2be2d0..9d94f511050 100644
--- a/ingestion/setup.py
+++ b/ingestion/setup.py
@@ -70,6 +70,7 @@ VERSIONS = {
     "sqlalchemy-bigquery": "sqlalchemy-bigquery>=1.2.2",
     "presidio-analyzer": "presidio-analyzer==2.2.358",
     "asammdf": "asammdf~=7.4.5",
+    "kafka-connect": "kafka-connect-py==0.10.11",
 }
 
 COMMONS = {
@@ -293,7 +294,7 @@ plugins: Dict[str, Set[str]] = {
         "thrift-sasl~=0.4",
     },
     "kafka": {*COMMONS["kafka"]},
-    "kafkaconnect": {"kafka-connect-py==0.10.11"},
+    "kafkaconnect": {VERSIONS["kafka-connect"]},
     "kinesis": {VERSIONS["boto3"]},
     "looker": {
         VERSIONS["looker-sdk"],
@@ -464,6 +465,7 @@ test = {
     "faker==37.1.0",  # The version needs to be fixed to prevent flaky tests!
     *plugins["exasol"],
     VERSIONS["opensearch"],
+    VERSIONS["kafka-connect"],
 }
 
 if sys.version_info >= (3, 9):
diff --git a/ingestion/src/metadata/ingestion/ometa/mixins/tests_mixin.py b/ingestion/src/metadata/ingestion/ometa/mixins/tests_mixin.py
index 38c3a2b2c6a..1e93b9ad535 100644
--- a/ingestion/src/metadata/ingestion/ometa/mixins/tests_mixin.py
+++ b/ingestion/src/metadata/ingestion/ometa/mixins/tests_mixin.py
@@ -253,7 +253,7 @@ class OMetaTestsMixin:
         }
 
         resp = self.client.get(
-            f"{self.get_suffix(TestCaseResult)}/{test_case_fqn}",
+            f"{self.get_suffix(TestCaseResult)}/{quote(test_case_fqn)}",
             params,
         )
 
diff --git a/ingestion/tests/integration/ometa/test_ometa_test_suite.py b/ingestion/tests/integration/ometa/test_ometa_test_suite.py
index 135ada0136a..7c26b566c76 100644
--- a/ingestion/tests/integration/ometa/test_ometa_test_suite.py
+++ b/ingestion/tests/integration/ometa/test_ometa_test_suite.py
@@ -128,6 +128,29 @@ class OMetaTestSuiteTest(TestCase):
             test_case_fqn="sample_data.ecommerce_db.shopify.dim_address.testCaseForIntegration",
         )
 
+        # Create test case with special characters in FQN to test URL encoding
+        cls.metadata.create_or_update(
+            CreateTestCaseRequest(
+                name=TestCaseEntityName("testCase:With/Special&Characters"),
+                entityLink=EntityLink(
+                    "<#E::table::sample_data.ecommerce_db.shopify.dim_address>"
+                ),
+                testDefinition=cls.test_definition.fullyQualifiedName,
+                parameterValues=[TestCaseParameterValue(name="foo", value="20")],
+            )
+        )
+
+        cls.metadata.add_test_case_results(
+            test_results=TestCaseResult(
+                timestamp=datetime_to_ts(datetime.now(timezone.utc)),
+                testCaseStatus=TestCaseStatus.Success,
+                result="Test Case with special chars Success",
+                sampleData=None,
+                testResultValue=[TestResultValue(name="foo", value="20")],
+            ),
+            test_case_fqn="sample_data.ecommerce_db.shopify.dim_address.testCase:With/Special&Characters",
+        )
+
     def test_get_or_create_test_suite(self):
         """test we get a test suite object"""
         test_suite = self.metadata.get_or_create_test_suite(
@@ -187,6 +210,22 @@ class OMetaTestSuiteTest(TestCase):
 
         assert res
 
+    def test_get_test_case_results_with_special_characters(self):
+        """test get test case results with special characters in FQN (: / &)"""
+        # This test validates that URL encoding works correctly for FQNs with special chars
+        res = self.metadata.get_test_case_results(
+            "sample_data.ecommerce_db.shopify.dim_address.testCase:With/Special&Characters",
+            get_beginning_of_day_timestamp_mill(),
+            get_end_of_day_timestamp_mill(),
+        )
+
+        assert (
+            res is not None
+        ), "Should fetch results for test case with special characters"
+        assert len(res) > 0, "Should have at least one result"
+        assert res[0].result == "Test Case with special chars Success"
+        assert res[0].testCaseStatus == TestCaseStatus.Success
+
     @classmethod
     def tearDownClass(cls) -> None:
         cls.metadata.delete_executable_test_suite(