Message Queue: Framing, Scale Math, and Architecture

This lesson opens Chapter 8: Distributed Message Queue at Hyperscale (Production System Blueprint 5: High-Throughput Append-Only Logs, Partition Sharding, and Cooperative Consumer Groups). It covers sections 1 to 3 of the chapter.

At a glance

1. The Scenario & Problem Framing

PROMPT:

"Design a horizontally scalable, distributed message queue and event streaming platform (like Apache Kafka, Apache Pulsar, or AWS Kinesis) capable of ingesting 10,000,000 messages per second with an average payload size of 1 KB (10 Gigabytes per second of raw write throughput). The system must guarantee at-least-once delivery, strict message ordering within individual entity partitions, sub-10ms produce latency (p99), and end-to-end consumer lag under 100ms. Provide full fault tolerance against arbitrary broker crashes, network partitions, and unannounced rolling deployments across consumer fleets without triggering multi-minute stop-the-world rebalance freezes."

In synchronous request-reply microservice architectures, communication is tight and brittle. If the Payment Service calls the Fraud Service and Fraud is slow, Payment's thread pool exhausts, HTTP connections back up, and the entire checkout flow cascades into an outage.

Distributed message queues break this temporal coupling. The producer emits an event, receives an acknowledgment in single-digit milliseconds, and moves on. The consuming services pull and process records asynchronously at their own sustainable rate.

However, designing an append-only event streaming system capable of ingesting 10 Gigabytes per second (80 Gbps of raw network traffic) represents one of the most brutal engineering challenges in distributed infrastructure:

1. The In-Memory Trap: Candidates often think: "We will store active messages in Redis or RabbitMQ RAM queues for fast retrieval." At 10 GB/sec, storing just 10 minutes of buffering requires 6 Terabytes of volatile RAM. If a downstream billing consumer halts for 2 hours, the cluster requires 72 TB of RAM and crashes with out-of-memory errors. Volatile memory cannot serve as a durable enterprise event log.
2. The JVM Context Switch & Buffer Copy Bottleneck: In traditional application servers, reading a file from disk and sending it over the network requires four context switches and two CPU memory copies (Disk -> OS Page Cache -> JVM Heap -> Socket Buffer -> NIC). At 10 GB/sec, CPU caches are instantly thrashed by memory copying, garbage collection cycles spike to 30 seconds, and throughput collapses.
3. The Global Ordering Fallacy: Product managers often say: "We need all events globally ordered." Global ordering requires a single partition, a single leader broker, and a single consumer thread. A single thread cannot process 10,000,000 events per second. Horizontal scalability requires sharding by entity key, preserving ordering only where business logic requires it.
4. The Stop-the-World Consumer Rebalance Disaster: When 500 consumer pods execute a rolling deployment, legacy eager rebalancing commands all consumers to revoke all partitions simultaneously. Event processing completely halts for 30 to 180 seconds, accumulating backlogs of tens of millions of unread messages and tripping SLA alarms across the company.

flowchart TB
    P["THE CORE PARADOX<br/><br/>A distributed message queue must achieve the raw sequential write speed of bare-metal disk hardware while providing linear consumer fan-out across hundreds of independent microservices without copying data into user-space RAM.<br/><br/>Staff Goal: Decouple physical append-only disk segments from network dispatch using Linux kernel zero-copy sendfile(), enforce quorum In-Sync Replicas (ISR) with monotonic leader epochs, and eliminate consumer rebalance blackouts via Cooperative Incremental Assignment."]

The L5 vs L6 Contrast — What the Evaluator Actually Hears

The Senior Candidate (L5): Immediately draws a Kafka cluster with topics and consumer groups, assuming that Kafka "gives us exactly-once delivery out of the box". When asked how to handle slow consumers, they suggest adding more consumer instances. When asked what happens if a broker crashes mid-write, they state that ZooKeeper handles it automatically. They treat Kafka as a magic black box, unaware of OS page cache writeback mechanics, filesystem sparse indexes, or the mathematical impossibility of end-to-end exactly-once processing without consumer-side idempotent deduplication.

The Staff Candidate (L6): Reframes the problem around reliability contracts, data lifecycle boundaries, and kernel I/O mechanics within the first 90 seconds: "A message queue is an immutable, append-only commit log sharded into independent partition units. At 10 GB/second write volume, we cannot afford JVM memory copies; we must rely entirely on Linux kernel zero-copy sendfile() streaming directly from the OS page cache to the network interface DMA buffer. Furthermore, 'exactly-once' is an application-level contract: the broker guarantees at-least-once durable append via In-Sync Replica quorums, while consumers enforce idempotency keys at their database write boundaries. Finally, we implement Cooperative Incremental Rebalancing so that rolling pod deployments never freeze unassigned partition streams."

2. Step 1: Commit to Intent & Scale Math

Before choosing technology stacks or drawing architecture boxes, an L6 Staff engineer establishes the core operational intent. The phrase "message queue" conceals three conflicting architectural paradigms:

The Three Incompatible Intents

1. High-Throughput Stream Pipeline (Apache Kafka / AWS Kinesis):

The Philosophy: Ingest millions of telemetry, tracking, and CDC events per second with maximum hardware saturation. Messages are immutable, persistent, and read by multiple independent consumer groups via monotonic offsets.

The Architectural Commit: Sequential append-only segment files, asynchronous OS page cache flushing, zero-copy network streaming, and pull-based consumer batching.

2. Low-Latency Individual Task Queue (RabbitMQ / AWS SQS):

The Philosophy: Discrete task distribution where individual workers consume one message at a time. Once a worker acknowledges a task, the message is permanently deleted from the queue.

The Architectural Commit: In-memory priority queues, broker-pushed messages, per-message tracking, and complex routing exchanges (AMQP). Fails at massive stream volume.

3. Financial Audit & State Replication Bus (Apache Pulsar / Apache BookKeeper):

The Philosophy: Strict linearizability and zero uncommitted data loss. Used for ledger updates, core banking transactions, and order state machines.

The Architectural Commit: Decoupled storage and compute layers, synchronous fsync multi-node barriers, and distributed consensus quorums.

STAFF MOVE

Senior: "We will configure Kafka with acks=all, enable idempotence, and set consumer commit to automatic."

Staff: "Before committing to client configurations: automatic offset commit creates either message loss or duplicate processing whenever a worker crashes mid-transaction. If auto.commit.interval.ms fires before the database transaction commits, a worker crash results in lost data. If the transaction commits before the auto-offset fires, a worker crash results in duplicate processing. For this 10M msg/sec pipeline, we must enforce manual synchronous offset commits coupled with transactional outboxes or consumer-side idempotent upserts (ON CONFLICT DO NOTHING)."

Why: Moving offset management from an automated background timer to an explicit boundary contract is the hallmark of Staff-level reliability design.

Capacity Planning & Scale Math

At 10,000,000 messages per second, sizing calculations determine network interface cards (NICs), disk I/O channels, memory page cache footprints, and partition allocations.

1. INBOUND INGESTION THROUGHPUT & BANDWIDTH

• Ingestion Rate: 10,000,000 messages/second.
• Average Message Size: 1 Kilobyte (including headers, payload, and metadata).

Raw Inbound Bandwidth:

Raw Ingestion = 10,000,000 msg/sec * 1 KB = 10,000,000 KB/sec ≈ 10 Gigabytes/second (80 Gbps)

Replication Factor (RF = 3): To guarantee survival against simultaneous rack failure and maintenance cycles, every partition has 1 leader and 2 follower replicas.

Internal Cluster Replication Traffic = 10 GB/sec * 2 Followers = 20 Gigabytes/second (160 Gbps)
Total Cluster Network Ingress/Egress = 10 GB/sec (Ingest) + 20 GB/sec (Replicate) = 30 GB/sec (240 Gbps)

2. STORAGE CAPACITY & TIERING MATH

Daily Ingestion Volume (Raw):

Daily Raw Volume = 10 GB/sec * 86,400 seconds = 864,000 Gigabytes ≈ 864 Terabytes / day!

With RF = 3: 864 TB * 3 = 2.592 Petabytes of raw storage per day!

7-Day Hot/Warm Retention:

7-Day Hot Storage = 2.592 PB/day * 7 days ≈ 18.14 Petabytes

The Storage Strategy: Sizing 18.1 PB of enterprise NVMe SSDs across physical brokers is prohibitively expensive (costing over $400,000/month). We implement Tiered Storage:

• Hot Tier (Local NVMe SSDs): Retain only the last 4 hours of data locally on brokers to serve real-time consumers from the OS page cache:

Hot Local Storage = (10 GB/sec * 3 replicas) * 14,400 sec ≈ 432 Terabytes total across cluster.

• Cold Tier (Object Store - AWS S3 / GCS): Segments older than 4 hours are sealed and asynchronously streamed to S3, reducing storage costs by 85% while retaining infinite replay capabilities!

3. SIZING THE BROKER FLEET & NETWORK INTERFACES

Hardware Profile: High-performance storage instances (e.g., AWS i3en.12xlarge or bare-metal equivalents) equipped with:

• 48 vCPUs, 384 GB ECC RAM.
• 4x 7.5 TB NVMe SSDs (30 TB raw local storage per node).
• 50 Gbps Dual-Port Network Interface Cards (NICs).

Effective Network Bandwidth per Broker: Sizing for 60% network saturation for safety headroom:

Safe Ingestion Bandwidth per Broker = 50 Gbps * 0.60 ≈ 30 Gbps ≈ 3.75 Gigabytes/second

Brokers Required:

Ingestion + Replication Bandwidth Needed = 30 Gigabytes/second
Minimum Broker Count = 30 GB/sec / (3.75 GB/sec/broker) = 8 brokers (absolute network minimum)
Storage Sizing for 432 TB Hot NVMe = 432 TB / (30 TB/broker) ≈ 15 brokers

Staff Cluster Sizing Call: To account for partition rebalancing, hot-spot skew, and node failure tolerance ((N + 2)), we deploy a 60-node broker cluster. Each broker handles ~166,000 msg/sec (166 MB/sec ingress), operating at a comfortable 25% CPU utilization and 35% network saturation.

4. MEMORY SIZING: THE 10-MINUTE PAGE CACHE WORKING SET

In Linux-based message brokers, Kafka does not cache messages in the JVM heap. It delegates 100% of read caching to the Linux OS Page Cache:

Active Consumer Working Set (Last 10 minutes of writes) = 10 GB/sec * 600 seconds = 6 Terabytes

Distributed across 60 brokers:

Page Cache per Broker = 6 TB / 60 brokers = 100 Gigabytes of RAM per host

With 384 GB of physical RAM per broker, over 250 GB per node is available exclusively for the OS Page Cache. Real-time consumers reading within 10 minutes of real-time achieve a 99.2% Page Cache Hit Rate, resulting in zero physical disk reads during normal streaming operations!

5. SIZING THE DOWNSTREAM CONSUMER FLEET (LITTLE'S LAW)

Using Little's Law (L = λ * W):

• Arrival Rate (λ): 10,000,000 messages/sec.
• Average Consumer Processing Latency (W): 15 milliseconds (JSON/Avro deserialization, validation, database commit).

Concurrent In-Flight Messages = 10,000,000 msg/sec * 0.015 sec = 150,000 concurrent messages

If each containerized consumer pod executes 50 concurrent worker threads:

Base Consumer Containers = 150,000 concurrent messages / 50 threads = 3,000 worker pods

Sizing for 70% target utilization with a 1.5x peak burst headroom requires 4,500 containerized consumer pods distributed across microservice consumer groups.

Key System Sizing Numbers Worth Memorizing

3. End-to-End System Architecture

The architecture decouples the producer ingestion network boundary from the physical disk append-only commit logs, leveraging OS Page Cache writeback and zero-copy consumer delivery:

Figure 8.1: End-to-End Hyperscale Distributed Message Queue Architecture

The Architecture Components in Depth

flowchart TB
    subgraph TITLE["HYPERSCALE DISTRIBUTED MESSAGE QUEUE RUNTIME ARCHITECTURE"]
        Prod["Producer Applications"]
        Cons["Consumer Microservice Fleet"]
        C1["COMPONENT 1: Producer SDK & Partition Hash Ring (Murmur2 Keying / Direct Leader Sockets)"]
        C2["COMPONENT 2: Broker Socket Layer & Epoll Reactor (Acceptor -> Processors -> IO Queue)"]
        C3["COMPONENT 3: Append-Only Commit Log & Sparse Segment Index (1 GB Segments + Memory Map)<br/>├─ Linux OS Page Cache Append (Sequential Writeback, Zero Synchronous fsync)<br/>└─ Sparse Binary Index (Every 4 KB Offset Indexing for O(1) Memory Binary Search)"]
        C4["COMPONENT 4: In-Sync Replicas (ISR) Quorum & High Watermark Advancement (acks=all)<br/>• High Watermark = min(LEO across active ISR)<br/>• Consumers strictly forbidden past HW"]
        C5["COMPONENT 5: Zero-Copy Network Dispatcher (Linux sendfile())<br/>• Direct Page Cache -> NIC Ring<br/>• 50 Gbps Line Rate @ <8% CPU"]
        C6["COMPONENT 6: KRaft Metadata Quorum & Group Coord<br/>• Raft Consensus Log for Topic/Partition State<br/>• consumer_offsets Internal Compacted Topic<br/>• Cooperative Sticky Rebalance Coordinator"]
        C7["COMPONENT 7: Tiered Cloud Storage<br/>• Sealed 1 GB Segments -> S3<br/>• Transparent HTTP Range Queries<br/>• Local NVMe Footprint Cut by 85%"]
        Prod -->|"Batch RecordAccumulator: 64 KB / 10ms"| C1
        C1 -->|"TCP NIO Frame over Direct TCP Socket"| C2
        C2 -->|"De-queued by KafkaRequestHandler Worker Thread"| C3
        C3 -->|"Follower FetchRequest"| C4
        C3 -->|"DMA Direct Memory"| C5
        C5 -->|"Zero-Copy sendfile"| Cons
        C4 -->|"Metadata, Offsets & Heartbeats"| C6
        C7 -->|"Historical Fetches"| C5
    end

Component 1: Producer Fleet & Partition Routing Hash Ring

Memory Buffering & Record Accumulator: The client producer SDK manages an in-memory double-buffered memory pool (RecordAccumulator). Incoming records are batched into byte buffers grouped by topic partition. Batches trigger network transmission when either batch.size = 65536 (64 KB) is reached or the linger.ms = 10ms aggregation timer expires, maximizing I/O efficiency without sacrificing latency.

Partition Key Hashing: For keyed records (e.g., account_id, order_id), the producer evaluates Murmur2(key) % Total_Partitions. This guarantees that all state mutations for a specific business entity are strictly serialized onto the same physical partition in total sequential order. For unkeyed records, the SDK uses modern uniform sticky partitioning to pack messages into full batches, preventing partition fragmentation.

Direct Broker Connection Topology: The producer queries cluster metadata upon bootstrap and maintains direct, persistent TCP sockets to the exact leader brokers for its target partitions. All intermediate L4/L7 load balancers and reverse proxies are eliminated from the data path, slashing network transit hops and preventing CPU bottlenecks.

Component 2: Broker Socket Layer & Non-Blocking Epoll Reactor

Reactor Threading Model: Implements an asymmetric Java NIO reactor pattern backed by the Linux epoll system call. A dedicated AcceptorThread binds to port 9092, accepts incoming client TCP connection handshakes, and assigns sockets to a pool of ProcessorThreads (num.network.threads = 16) using round-robin distribution.

Request Channel & Worker Thread Pools: Processor threads read network byte buffers, unpack framing headers, verify CRC32C checksums, and place decoded request objects into a bounded in-memory RequestChannel queue (queued.max.requests = 10000).

I/O Execution Fleet: A decoupled pool of worker threads (KafkaRequestHandler, num.io.threads = 32) pulls requests from the channel, writes records into partition log segments, coordinates replication quorums, and hands completed response buffers back to processor threads for asynchronous socket dispatch.

Component 3: Append-Only Commit Log & Sparse Segment Index

Sequential Disk Append Architecture: The core storage engine treats disk as an append-only log. For each partition, the broker manages a dedicated filesystem directory composed of sequential 1 GB immutable .log files paired with memory-mapped .index (offset to physical byte position) and .timeindex (timestamp to offset) sparse index files.

Page Cache Writeback & fsync Elimination: Writes invoke the Linux write() system call, writing directly into the kernel OS Page Cache. Kafka deliberately avoids synchronous per-batch fsync (flush.messages = Long.MAX_VALUE), delegating physical disk flushes to the background kernel pdflush daemon. This transforms write operations into pure memory appends, achieving 450 MB/sec throughput per NVMe drive.

Sparse Index Resolution: To keep memory usage negligible, Kafka writes an index entry only every 4 KB of log data (index.interval.bytes = 4096). To locate offset X, the broker performs an in-memory binary search on the memory-mapped .index file to identify the nearest base offset ≤ X, followed by a localized sequential scan of at most 4 KB of log data, executing queries in sub-microsecond time.

Component 4: In-Sync Replicas (ISR) Quorum Controller & High Watermark

Replication Quorum Mechanics: Every partition maintains 1 Leader and N-1 Followers (RF = 3). Followers execute high-performance ReplicaFetcherThreads that stream data from the Leader's local page cache via standard FetchRequest primitives.

High Watermark (HW) vs. Log End Offset (LEO):

• Log End Offset (LEO): The offset of the next record to be written in a broker's local partition log.
• High Watermark (HW): The minimum LEO across all active members of the In-Sync Replicas (ISR) set.

The Invariant: Consumers are strictly forbidden from reading past the High Watermark. The Leader advances the HW only after all ISR nodes replicate the batch into their local Page Cache, guaranteeing that no uncommitted or phantom records are ever exposed to downstream consumers.

Dynamic ISR Membership: If a follower broker fails to send a fetch heartbeat within replica.lag.time.max.ms = 30000, the leader drops it from the ISR and notifies the KRaft Controller.

Component 5: Zero-Copy Network Dispatcher (sendfile & DMA Acceleration)

Kernel User-Space Bypass: Standard file retrieval involves four context switches and three memory copies (Disk -> Page Cache -> User JVM Heap -> Socket Buffer -> NIC Ring). In high-throughput messaging, JVM heap copies introduce catastrophic garbage collection pauses.

Linux sendfile() System Call: Kafka invokes the sendfile() system call:

ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

The kernel transfers pages directly from the OS Page Cache into the network socket buffer, completely bypassing user space.

DMA Scatter-Gather Hardware Line Rate: With modern Network Interface Cards (NICs) supporting DMA scatter-gather, data copying to the socket buffer is eliminated entirely; only memory descriptors (physical addresses and lengths) are transferred. The NIC's Direct Memory Access controller reads bytes directly from OS Page Cache RAM and streams them over the physical Ethernet wire, saturating 50 Gbps line rates with less than 8% CPU utilization.

Component 6: KRaft Metadata Quorum & Group Coordinator

Integrated Raft Consensus Quorum (KRaft): Eliminates legacy ZooKeeper dependencies by running an event-driven Raft consensus cluster directly inside broker controller nodes. All cluster state changes (partition reassignments, leader elections, topic creations) are serialized as immutable records on the internal @metadata topic. Standby controllers maintain pre-materialized in-memory state, slashing partition failover times from 30 seconds down to under 250 milliseconds.

Group Coordinator & Consumer Offsets: Consumer groups elect a broker to act as their Group Coordinator. Committed read offsets are written to the internal, heavily compacted topic __consumer_offsets partitioned by hash(group_id).

Cooperative Sticky Rebalancer: Manages group member lifecycles (Stable, PreparingRebalance, CompletingRebalance). Uses two-phase incremental assignment to migrate only reassigned partitions, enabling rolling deployments with zero streaming downtime.

Component 7: Tiered Cloud Storage Offloader & Compactor (KIP-405)

Storage Hierarchy Decoupling: Separates hot low-latency data paths from cold archival storage. Active 1 GB segments reside on high-speed local NVMe SSDs to serve real-time consumers operating within a 4-hour window.

Asynchronous Object Storage Upload: Once a 1 GB log segment is sealed, the Remote Storage Manager (RSM) daemon asynchronously uploads the .log file and its sparse indexes to Amazon S3 or Google Cloud Storage. Upon multi-AZ upload confirmation, the local NVMe segment is safely deleted, retaining only a lightweight remote index pointer in local memory.

Transparent Range Queries: Catch-up batch workloads (e.g., daily machine learning feature extraction jobs reading 7-day-old logs) are transparently served by the broker fetching chunks from S3 via HTTP Range Requests (Range: bytes=8388608-16777215), streaming cold segments directly to consumers without exhausting NVMe capacity.

Step-by-Step Message Append & Zero-Copy Consume Flow

The following sequence details the complete lifecycle of a message from producer memory accumulation to broker page cache replication and zero-copy consumer delivery:

sequenceDiagram
    participant Prod as Producer SDK
    participant Epoll as Epoll Reactor
    participant RC as RequestChannel
    participant PC as Page Cache / NVMe
    participant ISR as ISR Follower
    participant KR as KRaft Coord
    participant Cons as Consumer Pod
    Prod->>Epoll: (1) Batch Accum (64KB/10ms)
    Epoll->>RC: (2) Parse & Enqueue Request
    RC->>PC: (3) Append Record, Advance LEO
    ISR->>PC: (4) Fetch Sync Replication
    PC->>ISR: (5) Advance HW, Quorum Confirmed
    PC-->>Prod: (6) Produce ACK (acks=all)
    Cons->>KR: (7) Fetch Zero-Copy
    PC-->>Cons: sendfile() DMA Stream
    Cons->>KR: (8) Commit Offset Sync

STEP 1: CLIENT RECORD ACCUMULATION & COMPRESSION

The producer application calls send(). The SDK appends the record into the RecordAccumulator buffer pool. It computes the target partition using Murmur2(key) % Total_Partitions. When the buffer reaches 64 KB or linger.ms = 10ms expires, the client compresses the entire buffer as a single block using Zstandard (zstd), cutting the byte payload by 75%, and dispatches the TCP frame directly to the partition leader's socket.

STEP 2: NON-BLOCKING SOCKET INGRESS & REQUEST ENQUEUEING

The broker's Linux epoll socket layer catches the incoming TCP packets. An AcceptorThread hands the socket channel to an active ProcessorThread. The processor thread reads the binary frame, validates the magic byte and CRC32C checksum, constructs a ProduceRequest object, and pushes it into the broker's central in-memory RequestChannel queue.

STEP 3: PAGE CACHE APPEND & LOCAL LEO ADVANCE

A KafkaRequestHandler I/O worker thread pulls the ProduceRequest from the queue. It validates topic authorization and appends the raw compressed byte stream directly into the Linux OS Page Cache of the active 1 GB log segment file via sequential write(). The broker increments its local Log End Offset (LEO). If the append crosses a 4 KB boundary, an entry is recorded in the memory-mapped sparse .index file.

STEP 4: FOLLOWER FETCH STREAMING & REPLICA LOG SYNC

Concurrently, follower brokers for the partition run continuous ReplicaFetcherThreads. A follower issues a FetchRequest referencing its own local LEO. The leader reads the requested byte range directly from its local Page Cache and streams it to the follower. The follower appends the bytes into its own local Page Cache and advances its local LEO.

STEP 5: QUORUM CONFIRMATION & HIGH WATERMARK ADVANCEMENT

The follower issues its subsequent FetchRequest, which informs the leader of the follower's updated LEO. Once all replicas in the active In-Sync Replicas (ISR) set have acknowledged replication, the leader advances the partition's High Watermark (HW) to match the committed offset. The records are now permanently committed and eligible for consumer reading.

STEP 6: PRODUCER QUORUM ACKNOWLEDGMENT DELIVERY

Because the producer configured acks=all, the leader constructs a ProduceResponse frame containing the base assigned offset, partition ID, and commit timestamp. The response is queued back onto the processor thread's output channel and written out over the persistent TCP socket to the client SDK in < 15 milliseconds.

STEP 7: CONSUMER ZERO-COPY STREAMING (SENDFILE)

A downstream consumer pod issues a FetchRequest(topic, partition, offset = 104200, max_bytes = 1MB). The broker validates that the requested offset is ≤ High Watermark. Rather than copying bytes into the JVM heap, the broker invokes the Linux sendfile() system call. The kernel DMA engine reads pages directly from the OS Page Cache into the Network Interface Card (NIC) ring buffer, streaming bytes over Ethernet directly to the consumer pod at line rate.

STEP 8: OFFSET COMMIT & COOPERATIVE STICKY REBALANCE

The consumer processes the batch and periodically commits its progress via commitSync() or commitAsync(). The Group Coordinator broker writes the committed offset to the internal compacted __consumer_offsets topic. If consumer pods are added or restarted, the CooperativeStickyAssignor executes an incremental two-phase rebalance, revoking only the migrating partitions while allowing all other consumer pods to stream uninterrupted.