Programming Fundamentals and Code Quality Questions

Encompasses core programming skills, data structures, basic algorithms, language fundamentals, and code quality practices. Expect proficiency with arrays, strings, lists, hash maps or dictionaries, sets, common collection operations, basic sorting and searching algorithms, and tradeoffs between data structures. Understand control flow, functions and modular design, classes and object oriented programming concepts including encapsulation, inheritance, and polymorphism, exception handling, file input and output, and common language idioms for mainstream interview languages such as Python, Java, and C plus plus. Emphasizes writing clean, readable, maintainable code: meaningful naming, modular functions, small interfaces, handling edge cases and errors, logging and documentation, simple testing and debugging strategies, and awareness of time and space complexity for common operations. Candidates should be able to implement correct solutions, follow language specific idioms where appropriate, and demonstrate attention to code quality and readability.

EasyTechnical

33 practiced

Given an array of integers, implement a Java method boolean containsDuplicate(int[] nums) that returns true if any value appears at least twice and false if every element is unique. Aim for O(n) time and O(n) additional space. Discuss trade-offs if input is sorted, immutable, or memory constrained.

Sample Answer

Approach: Use a HashSet to track seen values in one pass. If a number is already in the set, return true immediately. This yields O(n) time and O(n) extra space.

java

import java.util.HashSet;
import java.util.Set;

public class Solution {
    public boolean containsDuplicate(int[] nums) {
        // Null/short-circuit checks
        if (nums == null || nums.length < 2) return false;
        Set<Integer> seen = new HashSet<>();
        for (int x : nums) {
            if (!seen.add(x)) { // add returns false if x already present
                return true;
            }
        }
        return false;
    }
}

Key points:- Time: O(n) average (HashSet ops are O(1) amortized).- Space: O(n) worst-case (all unique).

Edge cases:- null array → false (or throw, per API contract)- empty or single-element → false- large arrays → watch memory/GC under SRE constraints

Trade-offs:- If input is sorted: duplicates are adjacent; can check neighbors in O(n) time and O(1) space (preferred).- If input is immutable: must not modify array; HashSet approach still works. If allowed to copy, sorting a copy gives O(n log n) time, O(n) space.- If memory-constrained: use sorting in-place (if mutable) to reduce extra space to O(1) but cost O(n log n). For severe memory limits and known value ranges, consider a bitmap or Bloom filter (probabilistic, possible false positives) or external streaming algorithms.

SRE note: consider production constraints — input size, memory/latency requirements, GC impact. Add input validation and defensive limits to avoid OOM in real systems.

MediumTechnical

27 practiced

Implement a RollingAverage class in Python that maintains the average of timestamped samples within a sliding window of window_seconds. API: add(timestamp: float, value: float) and average(now: float) -> float. Implementation should be O(1) amortized per update and O(k) memory where k is number of samples in window. Discuss bursty input handling and clock skew.

Sample Answer

Approach:Use a deque of (timestamp, value) plus a running sum. add() appends (or aggregates if same timestamp) and updates sum. average(now) first evicts outdated samples from left while subtracting from sum, then returns sum / count (or 0.0). Each sample is appended once and removed once — amortized O(1). Memory is O(k), where k = samples in window.

python

from collections import deque
from typing import Deque, Tuple

class RollingAverage:
    def __init__(self, window_seconds: float):
        self.window = window_seconds
        self.q: Deque[Tuple[float, float]] = deque()  # (timestamp, value)
        self.sum = 0.0
        self.count = 0

    def add(self, timestamp: float, value: float) -> None:
        # Aggregate if the tail has same timestamp to handle bursts cheaply
        if self.q and self.q[-1][0] == timestamp:
            ts, v = self.q.pop()
            v += value
            self.q.append((ts, v))
        else:
            self.q.append((timestamp, value))
            self.count += 1
        self.sum += value

    def average(self, now: float) -> float:
        cutoff = now - self.window
        # Evict old samples
        while self.q and self.q[0][0] <= cutoff:
            ts, v = self.q.popleft()
            self.sum -= v
            self.count -= 1
        return (self.sum / self.count) if self.count > 0 else 0.0

Key points and reasoning:- Running sum and deque let us update/evict in amortized O(1); each element is touched at most twice.- Aggregating same-timestamp entries reduces memory and work under bursts.- Use count for average; if you prefer weighted by number of raw points during aggregation, store (timestamp, sum, num_points).

Burst handling:- Bursts create many writes; aggregation by timestamp reduces queue length.- If bursts are extreme, consider bucketing timestamps into fixed sub-second bins (e.g., 100ms) to bound memory.

Clock skew and ordering:- This impl assumes incoming timestamps are roughly consistent. If timestamps can arrive out-of-order: - If slightly out-of-order, accept small inaccuracies or buffer briefly before admitting. - For large skew or malicious input, validate timestamps (e.g., reject future timestamps > tolerance) or use server monotonic time for now() and rely on event ingestion time instead.- For production SRE use, prefer ingest-time (monotonic) timestamps for eviction, or keep two modes: event-time (for accuracy) vs processing-time (for reliability).

MediumTechnical

28 practiced

Design and implement a cache in Python that supports get/put, per-key TTLs, and LRU eviction when capacity is exceeded. Ensure expired keys do not count toward capacity. Describe complexity and how you handle clock skew and concurrent access if used by multiple threads.

Sample Answer

Approach: use a hashmap for O(1) key lookup and a doubly-linked list to track LRU order. Store per-key expiry timestamps; on access or insertion remove expired entries so they don't count toward capacity. Use threading.Lock for concurrency; optionally use time.monotonic() to reduce clock-skew effects.

python

import time
import threading

class Node:
    def __init__(self, key, val, exp):
        self.key = key; self.val = val; self.exp = exp
        self.prev = self.next = None

class TTLCache:
    def __init__(self, capacity):
        self.cap = capacity
        self.map = {}        # key -> Node
        # dummy head/tail for doubly-linked list (most-recent after head)
        self.head = Node(None,None,None); self.tail = Node(None,None,None)
        self.head.next = self.tail; self.tail.prev = self.head
        self.lock = threading.Lock()

    def _now(self):
        return time.monotonic()

    def _remove(self, node):
        node.prev.next = node.next
        node.next.prev = node.prev
        del self.map[node.key]

    def _add_to_front(self, node):
        node.next = self.head.next; node.prev = self.head
        self.head.next.prev = node; self.head.next = node
        self.map[node.key] = node

    def _evict_if_needed(self):
        # remove expired first
        cur = self.tail.prev
        while cur is not self.head and len(self.map) >= self.cap:
            if cur.exp is not None and cur.exp <= self._now():
                prev = cur.prev
                self._remove(cur)
                cur = prev
            else:
                break
        # then evict LRU if still over capacity
        while len(self.map) > self.cap:
            lru = self.tail.prev
            self._remove(lru)

    def get(self, key):
        with self.lock:
            node = self.map.get(key)
            if not node:
                return None
            if node.exp is not None and node.exp <= self._now():
                self._remove(node)
                return None
            # move to front
            self._remove(node)
            self._add_to_front(node)
            return node.val

    def put(self, key, value, ttl=None):
        exp = None if ttl is None else self._now() + ttl
        with self.lock:
            if key in self.map:
                self._remove(self.map[key])
            node = Node(key, value, exp)
            self._add_to_front(node)
            self._evict_if_needed()

Key points:- O(1) get/put on average (hash + list ops).- Space O(capacity) for active (non-expired) entries.- Expired entries removed lazily on access/eviction to avoid scanning entire cache.- Use time.monotonic() to avoid system clock changes; for distributed systems, TTLs should be based on a single trusted clock (e.g., NTP-synced servers or centralized token with expiry).- For concurrency: coarse-grained lock here is simple and safe; for higher throughput use sharded locks or lock-free structures.

MediumTechnical

33 practiced

Implement a Python function follow(path: str, callback: Callable[[str], None]) that yields or callbacks new lines appended to a log file and correctly handles log rotation (file renamed and new file created). Explain handling of inode changes, truncation, and edge cases like rapid rotations.

Sample Answer

Approach: implement a tail -F style watcher that opens the file, seeks to end, reads new lines, and monitors file identity (inode + device) and size to detect rotation or truncation. On rotation (inode change) reopen the path and continue; on truncation (size smaller) seek to start of file.

python

import os
import time
from typing import Callable

def follow(path: str, callback: Callable[[str], None], poll_interval=0.2):
    """
    Follow a log file, calling callback for each new line.
    Handles rotation (inode change), truncation, and rapid rotations.
    """
    def stat(fpath):
        try:
            s = os.stat(fpath)
            return (s.st_ino, s.st_dev, s.st_size)
        except FileNotFoundError:
            return None

    fh = None
    last_id = None

    while True:
        cur_stat = stat(path)
        if cur_stat is None:
            # file missing (maybe between rotations) — wait and retry
            if fh:
                fh.close()
                fh = None
                last_id = None
            time.sleep(poll_interval)
            continue

        if fh is None:
            # open new file and, if it is the same inode as last seen, keep position; else start at 0
            fh = open(path, 'r', encoding='utf-8', errors='replace')
            ino = (cur_stat[0], cur_stat[1])
            if ino == last_id:
                # same file reopened (rare), seek to previous end
                fh.seek(cur_stat[2])
            else:
                # new file: start at end (like tail -F); if you want to read historical lines change to 0
                fh.seek(0, os.SEEK_END)
            last_id = ino

        # read lines
        line = fh.readline()
        if line:
            callback(line.rstrip('\n'))
            continue

        # no data: check for rotation or truncation
        latest = stat(path)
        if latest is None:
            # file disappeared
            fh.close()
            fh = None
            last_id = None
            time.sleep(poll_interval)
            continue

        latest_id = (latest[0], latest[1])
        if latest_id != last_id:
            # inode changed -> rotation. Keep reading remainder of old file first (if any), then switch.
            # close old handle and open new file
            fh.close()
            # If rotated file still exists under old name? Usually it's renamed; reopen path will be new file
            fh = open(path, 'r', encoding='utf-8', errors='replace')
            last_id = latest_id
            # start at beginning of new file
            fh.seek(0, os.SEEK_END)
            continue

        # same inode but size decreased -> truncation
        if latest[2] < fh.tell():
            fh.seek(0)
            continue

        time.sleep(poll_interval)

Key points:- Use inode+device to detect rotation reliably; file path can be replaced while old file still continues being written to.- Truncation (logtruncate) detected by file size < current position — seek to start.- Rapid rotations: we always compare stat each loop and reopen as soon as inode differs; small poll_interval reduces missed lines but increases CPU.- Keep file handle reading until EOF before switching to avoid losing final lines when rotated.Complexity: O(1) memory; CPU proportional to polling frequency. Edge cases: permission errors, very high write rates (use smaller interval or event-based APIs like inotify on Linux), multi-line writes (use buffering), encoding errors handled with errors='replace'.

HardTechnical

28 practiced

You inherit an on-call script with global state and no tests. Describe a concrete refactoring plan to improve testability and maintainability: identify seams, extract pure functions, introduce dependency injection for external effects, add unit tests and integration tests, and migrate configuration to structured formats. Include before/after code sketches and a rollout strategy.

Sample Answer

Situation: I inherited a single-file on-call script (~800 LOC) with global state, shelling out to APIs, and no tests. It periodically failed during incidents and was risky to change.

Plan (high level):- Identify seams: I/O (HTTP, file, subprocess), global mutable state, time, randomness, and config parsing.- Extract pure functions for computation/decision logic.- Introduce dependency injection for external effects (HTTP client, clock, runner).- Migrate ad-hoc config to structured YAML/JSON consumed by a loader.- Add unit tests for pure logic, mocked tests for injected deps, and a small integration test suite that runs in CI against a staging sandbox.- Rollout incrementally behind a feature flag and pair-program with on-call engineers.

Before (simplified):

python

# oncall_script.py (monolith)
import requests, os, time
STATE = {}
def run():
    resp = requests.get("https://pager/api")
    if resp.status_code==200:
        if not STATE.get("notified"):
            os.system("send-alert")
            STATE["notified"]=True
    time.sleep(60)

After (sketch):

python

# core.py - pure logic
def should_notify(response_body, state):
    # pure: no I/O
    return "urgent" in response_body and not state.get("notified")

# infra.py - injected effects
class Env:
    def __init__(self, http_client, runner, clock, storage, config): ...
    def fetch(self): return self.http_client.get(...)
    def run_alert(self): self.runner.run("send-alert")
    def now(self): return self.clock.now()

# app.py - wiring
def loop(env, state):
    resp = env.fetch()
    if should_notify(resp.body, state):
        env.run_alert()
        state["notified"]=True

Testing:- Unit tests for should_notify with varied inputs.- Mock Env for tests of loop to verify interactions (assert runner called).- Integration tests: containerized script using a fake pager API and ephemeral storage; verify end-to-end behavior.

Config migration:- Move from environment-embedded literals to YAML:

yaml

pager_url: "https://pager/api"
alert_command: "/usr/local/bin/send-alert"
poll_interval: 60

- Validate schema with jsonschema on startup.

Rollout strategy:1. Write tests and refactor in a feature branch; keep behavior identical.2. Add feature flag / toggle in config to switch between old and new runner.3. Run integration tests in CI, then deploy to staging; run both scripts in parallel with telemetry verifying identical actions.4. Gradual cutover: enable new path for low-impact teams, monitor SLOs and alert fidelity for 48–72 hours.5. Remove legacy code after 2 successful incident-free weeks; keep tests and add documentation + runbook updates.

Risk mitigation & collaboration:- Keep atomic commits: separate extraction of pure functions, then DI, then config migration.- Pair with on-call engineers; add logging and metrics in each step.- Ensure rollback path and maintain backward compatibility of config keys.

This produces safer, testable code, reduces blast radius of changes, and increases confidence during incidents.

Unlock Full Question Bank

Get access to hundreds of Programming Fundamentals and Code Quality interview questions and detailed answers.

Join thousands of developers preparing for their dream job.