diff --git a/recipe/simple_use_case/single_controller_demo.py b/recipe/simple_use_case/single_controller_demo.py
index 5592ce2c..af8689ab 100644
--- a/recipe/simple_use_case/single_controller_demo.py
+++ b/recipe/simple_use_case/single_controller_demo.py
@@ -53,13 +53,24 @@ def compute_loss(data1, _data2):
 
 def compute_reward(response_ids: torch.Tensor) -> TensorDict:
     """Simulate a reward model that scores each token position in the response.
+    Returns a TensorDict with a ``"rm_score"`` field whose shape matches
+    ``response_ids`` (i.e. one scalar per response token).
+    """
+    time.sleep(1)
+    reward = torch.randn_like(response_ids, dtype=torch.float32)
+
+    return TensorDict({"rm_score": reward}, batch_size=response_ids.size(0))
+
+
+def compute_advantage(rewards: torch.Tensor) -> TensorDict:
+    """Simulate the process of computing advantage.
 
     Returns a TensorDict with an ``"advantage"`` field whose shape matches
-    ``response_ids`` (i.e. one scalar per response token).
+    ``rewards`` (i.e. one scalar per reward).
     """
     time.sleep(1)
-    advantage = torch.randn_like(response_ids, dtype=torch.float32)
-    return TensorDict({"advantage": advantage}, batch_size=response_ids.size(0))
+    advantage = torch.randn_like(rewards, dtype=torch.float32)
+    return TensorDict({"advantage": advantage}, batch_size=rewards.size(0))
 
 
 class TrainingWorker:
@@ -89,7 +100,7 @@ def infer_batch(self, kv_meta: KVBatchMeta) -> KVBatchMeta:
         """Simulate forward-only inference"""
         # 1. Pull data from storage
         data = tq.kv_batch_get_by_meta(meta=kv_meta)
-        logger.info(f"compute_log_prob: got data {data}")
+        logger.info(f"infer_batch: got data {data}")
 
         # 2. Model forward
         output = compute_log_prob(data["prompt_ids"], data["response_ids"])
@@ -494,6 +505,13 @@ def fit(self):
             meta = tq.kv_batch_put(keys=meta.keys, partition_id=meta.partition_id, fields=reward_output)
             logger.info(f"demo reward KVBatchMeta: {meta}")
 
+            # ========================= Compute advantage =========================
+            meta.fields = ["response_ids", "ref_log_prob", "old_log_prob", "rm_score"]
+            advantage_data = tq.kv_batch_get_by_meta(meta=meta)
+            advantage_output = compute_advantage(advantage_data["rm_score"])
+            meta = tq.kv_batch_put(keys=meta.keys, partition_id=meta.partition_id, fields=advantage_output)
+            logger.info(f"demo advantage KVBatchMeta: {meta}")
+
             # ========================= Update actor =========================
             meta.fields = [
                 "input_ids",
diff --git a/transfer_queue/client.py b/transfer_queue/client.py
index 9b064c84..5f3a4e3f 100644
--- a/transfer_queue/client.py
+++ b/transfer_queue/client.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 
 import asyncio
-import logging
 import os
 import threading
 from typing import Any, Callable, Optional
@@ -32,6 +31,7 @@
     TransferQueueStorageManagerFactory,
 )
 from transfer_queue.utils.common import limit_pytorch_auto_parallel_threads
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.zmq_utils import (
     ZMQMessage,
     ZMQRequestType,
@@ -39,14 +39,7 @@
     with_zmq_socket,
 )
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 TQ_NUM_THREADS = int(os.environ.get("TQ_NUM_THREADS", 8))
 
diff --git a/transfer_queue/controller.py b/transfer_queue/controller.py
index 90304f7c..49aa2410 100644
--- a/transfer_queue/controller.py
+++ b/transfer_queue/controller.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 
 import copy
-import logging
 import os
 import time
 from collections import defaultdict
@@ -37,6 +36,7 @@
 )
 from transfer_queue.sampler import BaseSampler, SequentialSampler
 from transfer_queue.utils.enum_utils import TransferQueueRole
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.perf_utils import IntervalPerfMonitor
 from transfer_queue.utils.zmq_utils import (
     ZMQMessage,
@@ -48,14 +48,7 @@
     get_node_ip_address_raw,
 )
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler (for Ray Actor subprocess)
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 TQ_CONTROLLER_GET_METADATA_TIMEOUT = int(os.environ.get("TQ_CONTROLLER_GET_METADATA_TIMEOUT", 1))
 TQ_CONTROLLER_GET_METADATA_CHECK_INTERVAL = int(os.environ.get("TQ_CONTROLLER_GET_METADATA_CHECK_INTERVAL", 5))
diff --git a/transfer_queue/dataloader/streaming_dataloader.py b/transfer_queue/dataloader/streaming_dataloader.py
index 20a3b1d3..3b08cd26 100644
--- a/transfer_queue/dataloader/streaming_dataloader.py
+++ b/transfer_queue/dataloader/streaming_dataloader.py
@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 from typing import Optional
 
 import torch
@@ -22,15 +20,9 @@
 
 from transfer_queue.dataloader.streaming_dataset import StreamingDataset
 from transfer_queue.metadata import BatchMeta
+from transfer_queue.utils.logging_utils import get_logger
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 def _identity_collate_fn(data: tuple[TensorDict, BatchMeta]) -> tuple[TensorDict, BatchMeta]:
diff --git a/transfer_queue/dataloader/streaming_dataset.py b/transfer_queue/dataloader/streaming_dataset.py
index 62d5d86f..6c7165e6 100644
--- a/transfer_queue/dataloader/streaming_dataset.py
+++ b/transfer_queue/dataloader/streaming_dataset.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import os
 import time
 import uuid
@@ -25,19 +24,13 @@
 
 from transfer_queue.client import TransferQueueClient
 from transfer_queue.metadata import BatchMeta
+from transfer_queue.utils.logging_utils import get_logger
 
 TQ_STREAMING_DATASET_EMPTY_BATCH_SLEEP_INTERVAL = float(
     os.environ.get("TQ_STREAMING_DATASET_EMPTY_BATCH_SLEEP_INTERVAL", 1)
 )  # in seconds
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 class StreamingDataset(IterableDataset):
diff --git a/transfer_queue/interface.py b/transfer_queue/interface.py
index 31b5a232..66141b8f 100644
--- a/transfer_queue/interface.py
+++ b/transfer_queue/interface.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import math
 import os
 import subprocess
@@ -35,14 +34,14 @@
 from transfer_queue.sampler import BaseSampler
 from transfer_queue.storage.simple_backend import SimpleStorageUnit
 from transfer_queue.utils.common import get_placement_group
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.yuanrong_utils import (
     cleanup_yuanrong_resources,
     initialize_yuanrong_backend,
 )
 from transfer_queue.utils.zmq_utils import process_zmq_server_info
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+logger = get_logger(__name__)
 
 _TRANSFER_QUEUE_CLIENT: Any = None
 _TRANSFER_QUEUE_STORAGE: Any = None
diff --git a/transfer_queue/metadata.py b/transfer_queue/metadata.py
index 30ecc780..03784995 100644
--- a/transfer_queue/metadata.py
+++ b/transfer_queue/metadata.py
@@ -16,8 +16,6 @@
 import copy
 import dataclasses
 import itertools
-import logging
-import os
 from collections import defaultdict
 from dataclasses import dataclass
 from types import MappingProxyType
@@ -27,14 +25,9 @@
 import torch
 from tensordict import TensorDict
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+from transfer_queue.utils.logging_utils import get_logger
 
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 # ---------------------------------------------------------------------------
diff --git a/transfer_queue/storage/clients/mooncake_client.py b/transfer_queue/storage/clients/mooncake_client.py
index 6ab610ee..47a0c318 100644
--- a/transfer_queue/storage/clients/mooncake_client.py
+++ b/transfer_queue/storage/clients/mooncake_client.py
@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 import pickle
 from typing import Any, Optional
 
@@ -23,9 +21,9 @@
 
 from transfer_queue.storage.clients.base import TransferQueueStorageKVClient
 from transfer_queue.storage.clients.factory import StorageClientFactory
+from transfer_queue.utils.logging_utils import get_logger
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+logger = get_logger(__name__)
 
 MOONCAKE_STORE_IMPORTED: bool = True
 try:
diff --git a/transfer_queue/storage/clients/yuanrong_client.py b/transfer_queue/storage/clients/yuanrong_client.py
index fccd9a9b..bfd6cf49 100644
--- a/transfer_queue/storage/clients/yuanrong_client.py
+++ b/transfer_queue/storage/clients/yuanrong_client.py
@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 import struct
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
@@ -25,11 +23,11 @@
 
 from transfer_queue.storage.clients.base import TransferQueueStorageKVClient
 from transfer_queue.storage.clients.factory import StorageClientFactory
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.serial_utils import _decoder, _encoder
 from transfer_queue.utils.yuanrong_utils import find_reachable_host
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+logger = get_logger(__name__)
 
 
 YUANRONG_DATASYSTEM_IMPORTED: bool = True
diff --git a/transfer_queue/storage/managers/base.py b/transfer_queue/storage/managers/base.py
index 374db0a1..9f245a77 100644
--- a/transfer_queue/storage/managers/base.py
+++ b/transfer_queue/storage/managers/base.py
@@ -15,7 +15,6 @@
 
 import asyncio
 import itertools
-import logging
 import os
 import time
 import weakref
@@ -34,16 +33,10 @@
 
 from transfer_queue.metadata import BatchMeta, extract_field_schema
 from transfer_queue.storage.clients.factory import StorageClientFactory
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.zmq_utils import ZMQMessage, ZMQRequestType, ZMQServerInfo, create_zmq_socket
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 # ZMQ timeouts (in seconds) and retry configurations
 TQ_STORAGE_POLLER_TIMEOUT = int(os.environ.get("TQ_STORAGE_POLLER_TIMEOUT", 5))
diff --git a/transfer_queue/storage/managers/mooncake_manager.py b/transfer_queue/storage/managers/mooncake_manager.py
index a24ffafd..7b43ca71 100644
--- a/transfer_queue/storage/managers/mooncake_manager.py
+++ b/transfer_queue/storage/managers/mooncake_manager.py
@@ -13,16 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 from typing import Any
 
 from transfer_queue.storage.managers.base import KVStorageManager
 from transfer_queue.storage.managers.factory import TransferQueueStorageManagerFactory
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.zmq_utils import ZMQServerInfo
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+logger = get_logger(__name__)
 
 
 @TransferQueueStorageManagerFactory.register("MooncakeStore")
diff --git a/transfer_queue/storage/managers/simple_backend_manager.py b/transfer_queue/storage/managers/simple_backend_manager.py
index 005ca9a8..0771638b 100644
--- a/transfer_queue/storage/managers/simple_backend_manager.py
+++ b/transfer_queue/storage/managers/simple_backend_manager.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 
 import asyncio
-import logging
 import os
 import warnings
 from collections import defaultdict
@@ -30,6 +29,7 @@
 from transfer_queue.metadata import BatchMeta, extract_field_schema
 from transfer_queue.storage.managers.base import TransferQueueStorageManager
 from transfer_queue.storage.managers.factory import TransferQueueStorageManagerFactory
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.zmq_utils import (
     ZMQMessage,
     ZMQRequestType,
@@ -37,14 +37,7 @@
     with_zmq_socket,
 )
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT = int(os.environ.get("TQ_SIMPLE_STORAGE_SEND_RECV_TIMEOUT", 200))  # seconds
 
diff --git a/transfer_queue/storage/managers/yuanrong_manager.py b/transfer_queue/storage/managers/yuanrong_manager.py
index 69050acf..4c37cc73 100644
--- a/transfer_queue/storage/managers/yuanrong_manager.py
+++ b/transfer_queue/storage/managers/yuanrong_manager.py
@@ -13,22 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 from typing import Any
 
 from transfer_queue.storage.managers.base import KVStorageManager
 from transfer_queue.storage.managers.factory import TransferQueueStorageManagerFactory
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.zmq_utils import ZMQServerInfo
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 @TransferQueueStorageManagerFactory.register("Yuanrong")
diff --git a/transfer_queue/storage/simple_backend.py b/transfer_queue/storage/simple_backend.py
index 7f0a0012..bf334efc 100644
--- a/transfer_queue/storage/simple_backend.py
+++ b/transfer_queue/storage/simple_backend.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import os
 import time
 import weakref
@@ -26,6 +25,7 @@
 
 from transfer_queue.utils.common import limit_pytorch_auto_parallel_threads
 from transfer_queue.utils.enum_utils import TransferQueueRole
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.perf_utils import IntervalPerfMonitor
 from transfer_queue.utils.zmq_utils import (
     ZMQMessage,
@@ -37,14 +37,7 @@
     get_node_ip_address_raw,
 )
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler (for Ray Actor subprocess)
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 TQ_STORAGE_POLLER_TIMEOUT = int(os.environ.get("TQ_STORAGE_POLLER_TIMEOUT", 5))  # in seconds
 TQ_NUM_THREADS = int(os.environ.get("TQ_NUM_THREADS", 8))
diff --git a/transfer_queue/utils/common.py b/transfer_queue/utils/common.py
index a9d2b935..56f0dda8 100644
--- a/transfer_queue/utils/common.py
+++ b/transfer_queue/utils/common.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import os
 from contextlib import contextmanager
 from typing import Optional
@@ -22,16 +21,11 @@
 import ray
 import torch
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+from transfer_queue.utils.logging_utils import get_logger
 
-DEFAULT_TORCH_NUM_THREADS = torch.get_num_threads()
+logger = get_logger(__name__)
 
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+DEFAULT_TORCH_NUM_THREADS = torch.get_num_threads()
 
 
 def get_placement_group(num_ray_actors: int, num_cpus_per_actor: int = 1):
diff --git a/transfer_queue/utils/enum_utils.py b/transfer_queue/utils/enum_utils.py
index fd6a7842..929889da 100644
--- a/transfer_queue/utils/enum_utils.py
+++ b/transfer_queue/utils/enum_utils.py
@@ -13,18 +13,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 from enum import Enum
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+from transfer_queue.utils.logging_utils import get_logger
 
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 class ExplicitEnum(str, Enum):
diff --git a/transfer_queue/utils/logging_utils.py b/transfer_queue/utils/logging_utils.py
new file mode 100644
index 00000000..75e160d6
--- /dev/null
+++ b/transfer_queue/utils/logging_utils.py
@@ -0,0 +1,48 @@
+# Copyright 2025 Huawei Technologies Co., Ltd. All Rights Reserved.
+# Copyright 2025 The TransferQueue Team
+#
+# Licensed 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 logging
+import os
+
+_DEFAULT_LOG_FORMAT = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
+
+
+def get_logger(
+    name: str | None,
+    default_level: str = "WARNING",
+) -> logging.Logger:
+    """Create and configure a logger with consistent formatting.
+
+    Creates a logger with the specified name, sets its level based on the
+    TQ_LOGGING_LEVEL environment variable (or default), and adds a StreamHandler
+    if no handlers exist. This is particularly useful for Ray Actor subprocesses
+    that may not inherit logging configuration from the parent process.
+
+    Args:
+        name: The name for the logger, typically __name__.
+        default_level: The default logging level if TQ_LOGGING_LEVEL is not set.
+
+    Returns:
+        A configured logging.Logger instance.
+    """
+    logger = logging.getLogger(name)
+    logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", default_level))
+
+    if not logger.hasHandlers():
+        handler = logging.StreamHandler()
+        handler.setFormatter(logging.Formatter(_DEFAULT_LOG_FORMAT))
+        logger.addHandler(handler)
+
+    return logger
diff --git a/transfer_queue/utils/perf_utils.py b/transfer_queue/utils/perf_utils.py
index d7b375c3..2aa3d1e3 100644
--- a/transfer_queue/utils/perf_utils.py
+++ b/transfer_queue/utils/perf_utils.py
@@ -13,20 +13,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import os
 import time
 from collections import defaultdict
 from contextlib import contextmanager
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.INFO))
+from transfer_queue.utils.logging_utils import get_logger
 
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__, default_level="INFO")
 
 TQ_PERF_LOG_FLUSH_INTERVAL = float(os.environ.get("TQ_PERF_LOG_FLUSH_INTERVAL", 300))  # in seconds
 
diff --git a/transfer_queue/utils/serial_utils.py b/transfer_queue/utils/serial_utils.py
index 90f8592c..853b3a3e 100644
--- a/transfer_queue/utils/serial_utils.py
+++ b/transfer_queue/utils/serial_utils.py
@@ -17,8 +17,6 @@
 # This implementation is inspired by https://github.com/vllm-project/vllm/blob/main/vllm/v1/serial_utils.py
 
 
-import logging
-import os
 import pickle
 import warnings
 from collections.abc import Sequence
@@ -32,6 +30,8 @@
 from msgspec import msgpack
 from tensordict import TensorDictBase
 
+from transfer_queue.utils.logging_utils import get_logger
+
 CUSTOM_TYPE_PICKLE = 1
 CUSTOM_TYPE_CLOUDPICKLE = 2
 CUSTOM_TYPE_TENSOR = 3  # For tensor with buffer reference
@@ -43,8 +43,7 @@
 
 bytestr: TypeAlias = bytes | bytearray | memoryview | zmq.Frame
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
+logger = get_logger(__name__)
 
 # Ignore warnings about non-writable buffers from torch.frombuffer. Upper codes will ensure
 # the tensors are writable to users.
diff --git a/transfer_queue/utils/zmq_utils.py b/transfer_queue/utils/zmq_utils.py
index 2858e285..8af6aec3 100644
--- a/transfer_queue/utils/zmq_utils.py
+++ b/transfer_queue/utils/zmq_utils.py
@@ -13,8 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-import os
 import socket
 import time
 from dataclasses import dataclass
@@ -29,16 +27,10 @@
 from ray.util import get_node_ip_address
 
 from transfer_queue.utils.enum_utils import ExplicitEnum, TransferQueueRole
+from transfer_queue.utils.logging_utils import get_logger
 from transfer_queue.utils.serial_utils import decode, encode
 
-logger = logging.getLogger(__name__)
-logger.setLevel(os.getenv("TQ_LOGGING_LEVEL", logging.WARNING))
-
-# Ensure logger has a handler
-if not logger.hasHandlers():
-    handler = logging.StreamHandler()
-    handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(name)s - %(message)s"))
-    logger.addHandler(handler)
+logger = get_logger(__name__)
 
 
 bytestr: TypeAlias = bytes | bytearray | memoryview