Implement RotatingFileSink in hierarchy
Company: Bloomberg
Role: Data Engineer
Category: Software Engineering Fundamentals
Difficulty: medium
Interview Round: Technical Screen
You are given an existing Python 3.10 codebase with an abstract base class RecordSink (from abc import ABC, abstractmethod) that defines open() -> None, write(record: dict) -> None, and close() -> None. A Pipeline class instantiates a sink, calls sink.open() once, then for each record calls sink.write(record), and finally calls sink.close(); Pipeline must not be modified. Implement a new concrete RotatingFileSink that:
(
1) writes each record as a compact JSON line (UTF-
8) to disk,
(
2) rotates to a new file when either a maximum number of lines N is reached or total bytes exceed limit L, using filenames like prefix_00001.log, prefix_00002.log, etc.,
(
3) supports use as a context manager via __enter__/__exit__ while preserving existing open/close semantics,
(
4) is safe for concurrent write() calls from multiple threads using a lock,
(
5) raises a custom SinkClosedError if write() is invoked after close(), and
(
6) adheres to Liskov substitution so Pipeline works unchanged. Do not alter existing method signatures or the Pipeline. Provide the full class implementation and a brief unit test showing rotation and thread-safety.
Overview: This question evaluates a candidate's proficiency in Python object-oriented design, file I/O and JSON serialization, concurrency control via locking, context manager protocols, custom exception handling, and adherence to Liskov substitution when implementing a rotating file sink.
Read the full Bloomberg Data Engineer interview experience this question came from
Community answers
Answer by Xiaoming
import json
import os
import threading
from typing import Optional
from abc import ABC, abstractmethod
----- Existing ABC (for context only; in你实际工程中不要重复定义) -----
class RecordSink(ABC):
@abstractmethod
def open(self) -> None:
...
@abstractmethod
def write(self, record: dict) -> None:
...
@abstractmethod
def close(self) -> None:
...
----- Custom error -----
class SinkClosedError(RuntimeError):
"""Raised when write() is called after the sink has been closed."""
----- RotatingFileSink implementation -----
class RotatingFileSink(RecordSink):
"""
A file sink that writes compact JSON lines to rotating log files.
Filenames: _00001.log, _00002.log, ...
Rotate on:
max_lines (N): when number of lines in current file >= N
max_bytes (L): when current file size in bytes would exceed L on next write
Thread-safe write() using a lock.
Supports with context manager while preserving open/close semantics.
"""
def init(
self,
prefix: str,
max_lines: Optional[int] = None,
max_bytes: Optional[int] = None,
) -> None:
if max_lines is None and max_bytes is None:
rai