diff --git a/clients/client-python/gravitino/api/authorization/owner.py b/clients/client-python/gravitino/api/authorization/owner.py
new file mode 100644
index 00000000000..63d2b469d16
--- /dev/null
+++ b/clients/client-python/gravitino/api/authorization/owner.py
@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from abc import ABC, abstractmethod
+from enum import Enum
+
+
+class Owner(ABC):
+    """The interface of an owner. The owner represents the user or group who owns a metadata object."""
+
+    class Type(Enum):
+        """The type of the owner."""
+
+        USER = "USER"
+        GROUP = "GROUP"
+
+    @abstractmethod
+    def name(self) -> str:
+        """The name of the owner.
+
+        Returns:
+            str: The name of the owner.
+        """
+
+    @abstractmethod
+    def type(self) -> "Owner.Type":
+        """The type of the owner.
+
+        Returns:
+            Owner.Type: The type of the owner.
+        """
diff --git a/clients/client-python/gravitino/client/gravitino_client.py b/clients/client-python/gravitino/client/gravitino_client.py
index 862b4bce336..48fa0e76a33 100644
--- a/clients/client-python/gravitino/client/gravitino_client.py
+++ b/clients/client-python/gravitino/client/gravitino_client.py
@@ -19,12 +19,14 @@
 
 from typing import Dict, List, Optional
 
+from gravitino.api.authorization.owner import Owner
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
 from gravitino.api.job.job_handle import JobHandle
 from gravitino.api.job.job_template import JobTemplate
 from gravitino.api.job.job_template_change import JobTemplateChange
 from gravitino.api.job.supports_jobs import SupportsJobs
+from gravitino.api.metadata_object import MetadataObject
 from gravitino.api.tag.tag_operations import TagOperations
 from gravitino.auth.auth_data_provider import AuthDataProvider
 from gravitino.client.gravitino_client_base import GravitinoClientBase
@@ -330,3 +332,38 @@ def delete_tag(self, tag_name) -> bool:
             NoSuchMetalakeException: If the metalake does not exist.
         """
         return self.get_metalake().delete_tag(tag_name)
+
+    # Owner operations
+    def get_owner(self, metadata_object: MetadataObject) -> Optional[Owner]:
+        """Get the owner of a metadata object.
+
+        Args:
+            metadata_object: The metadata object to get the owner for.
+
+        Returns:
+            Optional[Owner]: The owner of the metadata object, or None if no owner is set.
+
+        Raises:
+            NoSuchMetadataObjectException: If the metadata object does not exist.
+            NotFoundException: If a related resource is not found.
+            MetalakeNotInUseException: If the metalake is not in use.
+        """
+        return self.get_metalake().get_owner(metadata_object)
+
+    def set_owner(
+        self, metadata_object: MetadataObject, owner_name: str, owner_type: Owner.Type
+    ) -> None:
+        """Set the owner of a metadata object.
+
+        Args:
+            metadata_object: The metadata object to set the owner for.
+            owner_name: The name of the owner.
+            owner_type: The type of the owner (USER or GROUP).
+
+        Raises:
+            NoSuchMetadataObjectException: If the metadata object does not exist.
+            NotFoundException: If a related resource is not found.
+            MetalakeNotInUseException: If the metalake is not in use.
+            UnsupportedOperationException: If the operation is not supported.
+        """
+        self.get_metalake().set_owner(metadata_object, owner_name, owner_type)
diff --git a/clients/client-python/gravitino/client/gravitino_metalake.py b/clients/client-python/gravitino/client/gravitino_metalake.py
index 431bfdd23b7..0789bde52c3 100644
--- a/clients/client-python/gravitino/client/gravitino_metalake.py
+++ b/clients/client-python/gravitino/client/gravitino_metalake.py
@@ -16,14 +16,16 @@
 # under the License.
 
 import logging
-from typing import Dict, List
+from typing import Dict, List, Optional
 
+from gravitino.api.authorization.owner import Owner
 from gravitino.api.catalog import Catalog
 from gravitino.api.catalog_change import CatalogChange
 from gravitino.api.job.job_handle import JobHandle
 from gravitino.api.job.job_template import JobTemplate
 from gravitino.api.job.job_template_change import JobTemplateChange
 from gravitino.api.job.supports_jobs import SupportsJobs
+from gravitino.api.metadata_object import MetadataObject
 from gravitino.api.tag.tag import Tag
 from gravitino.api.tag.tag_operations import TagOperations
 from gravitino.client.dto_converters import DTOConverters
@@ -40,6 +42,7 @@
 from gravitino.dto.requests.job_template_updates_request import (
     JobTemplateUpdatesRequest,
 )
+from gravitino.dto.requests.owner_set_request import OwnerSetRequest
 from gravitino.dto.requests.tag_create_request import TagCreateRequest
 from gravitino.dto.requests.tag_updates_request import TagUpdatesRequest
 from gravitino.dto.responses.catalog_list_response import CatalogListResponse
@@ -50,6 +53,8 @@
 from gravitino.dto.responses.job_response import JobResponse
 from gravitino.dto.responses.job_template_list_response import JobTemplateListResponse
 from gravitino.dto.responses.job_template_response import JobTemplateResponse
+from gravitino.dto.responses.owner_response import OwnerResponse
+from gravitino.dto.responses.set_response import SetResponse
 from gravitino.dto.responses.tag_response import (
     TagListResponse,
     TagNamesListResponse,
@@ -57,6 +62,7 @@
 )
 from gravitino.exceptions.handlers.catalog_error_handler import CATALOG_ERROR_HANDLER
 from gravitino.exceptions.handlers.job_error_handler import JOB_ERROR_HANDLER
+from gravitino.exceptions.handlers.owner_error_handler import OWNER_ERROR_HANDLER
 from gravitino.exceptions.handlers.tag_error_handler import TAG_ERROR_HANDLER
 from gravitino.rest.rest_utils import encode_string
 from gravitino.utils.http_client import HTTPClient
@@ -82,6 +88,7 @@ class GravitinoMetalake(
     API_METALAKES_CATALOGS_PATH = "api/metalakes/{}/catalogs/{}"
     API_METALAKES_JOB_TEMPLATES_PATH = "api/metalakes/{}/jobs/templates"
     API_METALAKES_JOB_RUNS_PATH = "api/metalakes/{}/jobs/runs"
+    API_METALAKES_OWNERS_PATH = "api/metalakes/{}/owners/{}"
     API_METALAKES_TAG_PATH = "api/metalakes/{}/tags/{}"
     API_METALAKES_TAGS_PATH = "api/metalakes/{}/tags"
 
@@ -704,3 +711,57 @@ def delete_tag(self, tag_name) -> bool:
         drop_response.validate()
 
         return drop_response.dropped()
+
+    #########
+    # Owner operations
+    #########
+    def get_owner(self, metadata_object: MetadataObject) -> Optional[Owner]:
+        """Get the owner of a metadata object.
+
+        Args:
+            metadata_object: The metadata object to get the owner for.
+
+        Returns:
+            Optional[Owner]: The owner of the metadata object, or None if no owner is set.
+
+        Raises:
+            NoSuchMetadataObjectException: If the metadata object does not exist.
+            NotFoundException: If a related resource is not found.
+            MetalakeNotInUseException: If the metalake is not in use.
+        """
+        url = self.API_METALAKES_OWNERS_PATH.format(
+            encode_string(self.name()),
+            f"{metadata_object.type().value}/{encode_string(metadata_object.full_name())}",
+        )
+        response = self.rest_client.get(url, error_handler=OWNER_ERROR_HANDLER)
+        resp = OwnerResponse.from_json(response.body, infer_missing=True)
+        resp.validate()
+        return resp.owner()
+
+    def set_owner(
+        self, metadata_object: MetadataObject, owner_name: str, owner_type: Owner.Type
+    ) -> None:
+        """Set the owner of a metadata object.
+
+        Args:
+            metadata_object: The metadata object to set the owner for.
+            owner_name: The name of the owner.
+            owner_type: The type of the owner (USER or GROUP).
+
+        Raises:
+            NoSuchMetadataObjectException: If the metadata object does not exist.
+            NotFoundException: If a related resource is not found.
+            MetalakeNotInUseException: If the metalake is not in use.
+            UnsupportedOperationException: If the operation is not supported.
+        """
+        url = self.API_METALAKES_OWNERS_PATH.format(
+            encode_string(self.name()),
+            f"{metadata_object.type().value}/{encode_string(metadata_object.full_name())}",
+        )
+        req = OwnerSetRequest(owner_name, owner_type)
+        req.validate()
+        response = self.rest_client.put(
+            url, json=req, error_handler=OWNER_ERROR_HANDLER
+        )
+        set_resp = SetResponse.from_json(response.body, infer_missing=True)
+        set_resp.validate()
diff --git a/clients/client-python/gravitino/dto/authorization/__init__.py b/clients/client-python/gravitino/dto/authorization/__init__.py
new file mode 100644
index 00000000000..26a4e378cf4
--- /dev/null
+++ b/clients/client-python/gravitino/dto/authorization/__init__.py
@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from gravitino.dto.authorization.owner_dto import OwnerDTO
diff --git a/clients/client-python/gravitino/dto/authorization/owner_dto.py b/clients/client-python/gravitino/dto/authorization/owner_dto.py
new file mode 100644
index 00000000000..a68cd6abb14
--- /dev/null
+++ b/clients/client-python/gravitino/dto/authorization/owner_dto.py
@@ -0,0 +1,37 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from dataclasses import dataclass, field
+
+from dataclasses_json import config, dataclass_json
+
+from gravitino.api.authorization.owner import Owner
+
+
+@dataclass_json
+@dataclass
+class OwnerDTO(Owner):
+    """Represents an Owner Data Transfer Object (DTO)."""
+
+    _name: str = field(metadata=config(field_name="name"))
+    _type: Owner.Type = field(metadata=config(field_name="type"))
+
+    def name(self) -> str:
+        return self._name
+
+    def type(self) -> Owner.Type:
+        return self._type
diff --git a/clients/client-python/gravitino/dto/requests/owner_set_request.py b/clients/client-python/gravitino/dto/requests/owner_set_request.py
new file mode 100644
index 00000000000..b315db9546d
--- /dev/null
+++ b/clients/client-python/gravitino/dto/requests/owner_set_request.py
@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from dataclasses import dataclass, field
+
+from dataclasses_json import config, dataclass_json
+
+from gravitino.api.authorization.owner import Owner
+from gravitino.exceptions.base import IllegalArgumentException
+from gravitino.rest.rest_message import RESTRequest
+
+
+@dataclass_json
+@dataclass
+class OwnerSetRequest(RESTRequest):
+    """Represents a request to set an owner for a metadata object."""
+
+    _name: str = field(metadata=config(field_name="name"))
+    _type: Owner.Type = field(metadata=config(field_name="type"))
+
+    def __init__(self, name: str, owner_type: Owner.Type):
+        self._name = name
+        self._type = owner_type
+
+    def validate(self) -> None:
+        if not self._name:
+            raise IllegalArgumentException('"name" field is required')
+        if self._type is None:
+            raise IllegalArgumentException('"type" field is required')
diff --git a/clients/client-python/gravitino/dto/responses/owner_response.py b/clients/client-python/gravitino/dto/responses/owner_response.py
new file mode 100644
index 00000000000..71aaff2ea79
--- /dev/null
+++ b/clients/client-python/gravitino/dto/responses/owner_response.py
@@ -0,0 +1,45 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+from dataclasses_json import config
+
+from gravitino.dto.authorization.owner_dto import OwnerDTO
+from gravitino.dto.responses.base_response import BaseResponse
+from gravitino.exceptions.base import IllegalArgumentException
+
+
+@dataclass
+class OwnerResponse(BaseResponse):
+    """Represents a response for the get owner operation."""
+
+    _owner: Optional[OwnerDTO] = field(
+        default=None, metadata=config(field_name="owner")
+    )
+
+    def owner(self) -> Optional[OwnerDTO]:
+        return self._owner
+
+    def validate(self) -> None:
+        super().validate()
+        if self._owner is not None:
+            if not self._owner.name():
+                raise IllegalArgumentException("owner name must not be empty")
+            if self._owner.type() is None:
+                raise IllegalArgumentException("owner type must not be None")
diff --git a/clients/client-python/gravitino/dto/responses/set_response.py b/clients/client-python/gravitino/dto/responses/set_response.py
new file mode 100644
index 00000000000..23a9b8ebdbb
--- /dev/null
+++ b/clients/client-python/gravitino/dto/responses/set_response.py
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from dataclasses import dataclass, field
+
+from dataclasses_json import config
+
+from gravitino.dto.responses.base_response import BaseResponse
+
+
+@dataclass
+class SetResponse(BaseResponse):
+    """Represents a response for a set operation."""
+
+    _set: bool = field(metadata=config(field_name="set"))
+
+    def set(self) -> bool:
+        return self._set
diff --git a/clients/client-python/gravitino/exceptions/handlers/owner_error_handler.py b/clients/client-python/gravitino/exceptions/handlers/owner_error_handler.py
new file mode 100644
index 00000000000..bda819cd82e
--- /dev/null
+++ b/clients/client-python/gravitino/exceptions/handlers/owner_error_handler.py
@@ -0,0 +1,58 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from gravitino.constants.error import ErrorConstants
+from gravitino.exceptions.base import (
+    IllegalArgumentException,
+    MetalakeNotInUseException,
+    NoSuchMetadataObjectException,
+    NotFoundException,
+    UnsupportedOperationException,
+)
+from gravitino.exceptions.handlers.rest_error_handler import RestErrorHandler
+
+
+class OwnerErrorHandler(RestErrorHandler):
+    """Error handler specific to Owner operations."""
+
+    def handle(self, error_response) -> None:
+        error_message = error_response.format_error_message()
+        code = error_response.code()
+        exception_type = error_response.type()
+
+        if code == ErrorConstants.ILLEGAL_ARGUMENTS_CODE:
+            raise IllegalArgumentException(error_message)
+
+        if code == ErrorConstants.NOT_FOUND_CODE:
+            if exception_type == NoSuchMetadataObjectException.__name__:
+                raise NoSuchMetadataObjectException(error_message)
+
+            raise NotFoundException(error_message)
+
+        if code == ErrorConstants.UNSUPPORTED_OPERATION_CODE:
+            raise UnsupportedOperationException(error_message)
+
+        if code == ErrorConstants.NOT_IN_USE_CODE:
+            raise MetalakeNotInUseException(error_message)
+
+        if code == ErrorConstants.INTERNAL_ERROR_CODE:
+            raise RuntimeError(error_message)
+
+        super().handle(error_response)
+
+
+OWNER_ERROR_HANDLER = OwnerErrorHandler()
diff --git a/clients/client-python/tests/unittests/test_owner.py b/clients/client-python/tests/unittests/test_owner.py
new file mode 100644
index 00000000000..df20303ef27
--- /dev/null
+++ b/clients/client-python/tests/unittests/test_owner.py
@@ -0,0 +1,253 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF 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.
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from gravitino import GravitinoClient
+from gravitino.api.authorization.owner import Owner
+from gravitino.api.metadata_object import MetadataObject
+from gravitino.dto.authorization.owner_dto import OwnerDTO
+from gravitino.dto.metadata_object_dto import MetadataObjectDTO
+from gravitino.dto.requests.owner_set_request import OwnerSetRequest
+from gravitino.dto.responses.error_response import ErrorResponse
+from gravitino.dto.responses.owner_response import OwnerResponse
+from gravitino.dto.responses.set_response import SetResponse
+from gravitino.exceptions.base import (
+    IllegalArgumentException,
+    MetalakeNotInUseException,
+    NoSuchMetadataObjectException,
+    NotFoundException,
+    UnsupportedOperationException,
+)
+from gravitino.exceptions.handlers.owner_error_handler import OWNER_ERROR_HANDLER
+from tests.unittests import mock_base
+
+
+@mock_base.mock_data
+class TestOwner(unittest.TestCase):
+    _metalake_name: str = "metalake_demo"
+
+    def test_get_owner(self, *mock_method) -> None:
+        owner_resp = OwnerResponse(0, {"name": "alice", "type": "USER"})
+        json_str = owner_resp.to_json()
+        mock_resp = mock_base.mock_http_response(json_str)
+        client = GravitinoClient(
+            uri="http://localhost:8090",
+            metalake_name=self._metalake_name,
+            check_version=False,
+        )
+
+        metadata_object = (
+            MetadataObjectDTO.builder()
+            .type(MetadataObject.Type.CATALOG)
+            .full_name("test_catalog")
+            .build()
+        )
+
+        with patch(
+            "gravitino.utils.http_client.HTTPClient.get",
+            return_value=mock_resp,
+        ):
+            owner = client.get_owner(metadata_object)
+            self.assertIsNotNone(owner)
+            self.assertEqual("alice", owner.name())
+            self.assertEqual(Owner.Type.USER, owner.type())
+
+    def test_get_owner_none(self, *mock_method) -> None:
+        owner_resp = OwnerResponse(0, None)
+        json_str = owner_resp.to_json()
+        mock_resp = mock_base.mock_http_response(json_str)
+        client = GravitinoClient(
+            uri="http://localhost:8090",
+            metalake_name=self._metalake_name,
+            check_version=False,
+        )
+
+        metadata_object = (
+            MetadataObjectDTO.builder()
+            .type(MetadataObject.Type.CATALOG)
+            .full_name("test_catalog")
+            .build()
+        )
+
+        with patch(
+            "gravitino.utils.http_client.HTTPClient.get",
+            return_value=mock_resp,
+        ):
+            owner = client.get_owner(metadata_object)
+            self.assertIsNone(owner)
+
+    def test_set_owner(self, *mock_method) -> None:
+        set_resp = SetResponse(0, True)
+        json_str = set_resp.to_json()
+        mock_resp = mock_base.mock_http_response(json_str)
+        client = GravitinoClient(
+            uri="http://localhost:8090",
+            metalake_name=self._metalake_name,
+            check_version=False,
+        )
+
+        metadata_object = (
+            MetadataObjectDTO.builder()
+            .type(MetadataObject.Type.CATALOG)
+            .full_name("test_catalog")
+            .build()
+        )
+
+        with patch(
+            "gravitino.utils.http_client.HTTPClient.put",
+            return_value=mock_resp,
+        ) as mock_put:
+            client.set_owner(metadata_object, "alice", Owner.Type.USER)
+            mock_put.assert_called_once()
+
+    def test_set_owner_returns_false(self, *mock_method) -> None:
+        set_resp = SetResponse(0, False)
+        json_str = set_resp.to_json()
+        mock_resp = mock_base.mock_http_response(json_str)
+        client = GravitinoClient(
+            uri="http://localhost:8090",
+            metalake_name=self._metalake_name,
+            check_version=False,
+        )
+
+        metadata_object = (
+            MetadataObjectDTO.builder()
+            .type(MetadataObject.Type.CATALOG)
+            .full_name("test_catalog")
+            .build()
+        )
+
+        with patch(
+            "gravitino.utils.http_client.HTTPClient.put",
+            return_value=mock_resp,
+        ) as mock_put:
+            client.set_owner(metadata_object, "alice", Owner.Type.USER)
+            mock_put.assert_called_once()
+
+    def test_get_owner_with_group(self, *mock_method) -> None:
+        owner_resp = OwnerResponse(0, {"name": "admin_group", "type": "GROUP"})
+        json_str = owner_resp.to_json()
+        mock_resp = mock_base.mock_http_response(json_str)
+        client = GravitinoClient(
+            uri="http://localhost:8090",
+            metalake_name=self._metalake_name,
+            check_version=False,
+        )
+
+        metadata_object = (
+            MetadataObjectDTO.builder()
+            .type(MetadataObject.Type.SCHEMA)
+            .full_name("test_catalog.test_schema")
+            .build()
+        )
+
+        with patch(
+            "gravitino.utils.http_client.HTTPClient.get",
+            return_value=mock_resp,
+        ):
+            owner = client.get_owner(metadata_object)
+            self.assertIsNotNone(owner)
+            self.assertEqual("admin_group", owner.name())
+            self.assertEqual(Owner.Type.GROUP, owner.type())
+
+
+class TestOwnerErrorHandler(unittest.TestCase):
+    def test_illegal_arguments(self):
+        with self.assertRaises(IllegalArgumentException):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(
+                    IllegalArgumentException, "mock error"
+                )
+            )
+
+    def test_not_found_metadata_object(self):
+        with self.assertRaises(NoSuchMetadataObjectException):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(
+                    NoSuchMetadataObjectException, "mock error"
+                )
+            )
+
+    def test_not_found_generic(self):
+        with self.assertRaises(NotFoundException):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(NotFoundException, "mock error")
+            )
+
+    def test_unsupported_operation(self):
+        with self.assertRaises(UnsupportedOperationException):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(
+                    UnsupportedOperationException, "mock error"
+                )
+            )
+
+    def test_not_in_use(self):
+        with self.assertRaises(MetalakeNotInUseException):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(
+                    MetalakeNotInUseException, "mock error"
+                )
+            )
+
+    def test_internal_error(self):
+        with self.assertRaises(RuntimeError):
+            OWNER_ERROR_HANDLER.handle(
+                ErrorResponse.generate_error_response(RuntimeError, "mock error")
+            )
+
+
+class TestOwnerSetRequestValidation(unittest.TestCase):
+    def test_validate_empty_name(self):
+        req = OwnerSetRequest.__new__(OwnerSetRequest)
+        req._name = ""
+        req._type = Owner.Type.USER
+        with self.assertRaises(ValueError):
+            req.validate()
+
+    def test_validate_none_type(self):
+        req = OwnerSetRequest.__new__(OwnerSetRequest)
+        req._name = "alice"
+        req._type = None
+        with self.assertRaises(ValueError):
+            req.validate()
+
+    def test_validate_success(self):
+        req = OwnerSetRequest("alice", Owner.Type.USER)
+        req.validate()
+
+
+class TestOwnerResponseValidation(unittest.TestCase):
+    def test_validate_owner_with_empty_name(self):
+        owner_dto = OwnerDTO(_name="", _type=Owner.Type.USER)
+        resp = OwnerResponse(0, owner_dto)
+        with self.assertRaises(ValueError):
+            resp.validate()
+
+    def test_validate_owner_with_none_type(self):
+        owner_dto = OwnerDTO(_name="alice", _type=None)
+        resp = OwnerResponse(0, owner_dto)
+        with self.assertRaises(ValueError):
+            resp.validate()
+
+    def test_validate_no_owner(self):
+        resp = OwnerResponse(0, None)
+        resp.validate()