HBase Row Key Hotspots — Why One RegionServer Hit 100% CPU

package io.thecodeforge.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Production-grade HBase client demonstrating the read/write path.
 *
 * Key production rules:
 *   1. Connection is heavyweight and thread-safe — create once, share across threads
 *   2. Table is lightweight — obtain per-operation, close after use
 *   3. Always set retries and pause — transient RegionServer failures are normal
 *   4. Always set scan caching — default of 1 row per RPC is catastrophically slow
 */
public class HBaseArchitectureClient implements AutoCloseable {

    private final Connection connection;
    private final Admin admin;

    public HBaseArchitectureClient(String zkQuorum, int zkPort) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkQuorum);
        conf.setInt("hbase.zookeeper.property.clientPort", zkPort);

        // Retry on transient failures: up to 3 retries with 1s pause between
        conf.setInt("hbase.client.retries.number", 3);
        conf.setInt("hbase.client.pause", 1000);

        // Operation timeout — fail fast rather than queue behind a slow RegionServer
        conf.setInt("hbase.rpc.timeout", 5000);         // 5s per individual RPC
        conf.setInt("hbase.client.operation.timeout", 15000); // 15s total including retries

        // Connection is thread-safe and maintains a connection pool to each RegionServer
        // One Connection per application process is the correct pattern
        this.connection = ConnectionFactory.createConnection(conf);
        this.admin = connection.getAdmin();
    }

    /**
     * Write path: Put -> WAL -> MemStore -> (eventually) HFile on HDFS
     *
     * The WAL write happens before the client receives acknowledgment.
     * If the RegionServer crashes before MemStore flush, WAL replay recovers the data.
     * No acknowledged write is ever lost, even in a hard crash scenario.
     */
    public void writeRow(String tableName, String rowKey,
                         String columnFamily, String qualifier, byte[] value)
            throws IOException {

        // Table is lightweight — safe to create per-operation; always close after use
        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(
                Bytes.toBytes(columnFamily),
                Bytes.toBytes(qualifier),
                value
            );
            // setDurability controls WAL behaviour:
            //   SYNC_WAL (default)  — fsync WAL before ack — safest
            //   ASYNC_WAL           — buffer WAL writes — faster, small loss window
            //   SKIP_WAL            — no WAL — fastest, data loss on crash
            put.setDurability(Durability.SYNC_WAL);
            table.put(put);
        }
    }

    /**
     * Read path: Get -> MemStore -> BlockCache -> Bloom filter -> HFile block index -> HDFS
     *
     * Sub-10ms latency when the row is in MemStore (unflushed write) or BlockCache.
     * Cold reads (data not in any cache) depend on HDFS read latency — typically 20-50ms.
     */
    public byte[] readRow(String tableName, String rowKey,
                          String columnFamily, String qualifier)
            throws IOException {

        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));

            // setCacheBlocks(true) is the default — hot rows stay in BlockCache
            // Set to false for analytical/scan workloads to avoid evicting hot data
            get.setCacheBlocks(true);

            Result result = table.get(get);

            if (result.isEmpty()) {
                return null;  // Row does not exist — always check before unwrapping
            }

            return result.getValue(
                Bytes.toBytes(columnFamily),
                Bytes.toBytes(qualifier)
            );
        }
    }

    /**
     * Scan with explicit start/stop row — the correct production scan pattern.
     *
     * Never scan without start/stop rows unless you intend a full table scan.
     * Full table scans block RegionServer threads for minutes and evict hot
     * BlockCache data, degrading point lookup latency for all other clients.
     */
    public void scanRange(String tableName, String startKey, String stopKey)
            throws IOException {

        try (Table table = connection.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan()
                .withStartRow(Bytes.toBytes(startKey), true)   // inclusive
                .withStopRow(Bytes.toBytes(stopKey), false)    // exclusive
                .setCaching(100)             // 100 rows per RPC — reduces round trips
                .setMaxResultSize(2 * 1024 * 1024L)  // 2MB max per RPC response
                .setCacheBlocks(false);      // scans read sequentially — don't pollute BlockCache

            try (ResultScanner scanner = table.getScanner(scan)) {
                for (Result result : scanner) {
                    byte[] rowKey = result.getRow();
                    // process each row here — scanner handles region boundaries transparently
                    _ = rowKey; // suppress unused warning in example
                }
            }
        }
    }

    @Override
    public void close() throws IOException {
        admin.close();
        connection.close();
    }
}

package io.thecodeforge.hbase;

import org.apache.hadoop.hbase.util.Bytes;

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;

/**
 * Production row key design patterns for HBase.
 *
 * Each pattern makes a different trade-off between write distribution,
 * range scan efficiency, and deterministic lookup. Choose based on
 * your dominant access pattern — not based on what looks simplest.
 *
 * Decision guide:
 *   Write-heavy + rare scans    → salt prefix
 *   Time-series + recent reads  → reversed timestamp
 *   Read-heavy + point lookups  → hash prefix
 *   Multi-dimensional queries   → composite key
 */
public class RowKeyDesign {

    /**
     * Pattern 1: Salt prefix for write distribution.
     *
     * Distributes writes uniformly across all regions.
     * Trade-off: range scans on the natural key require querying all regions
     * in parallel and merging results in the client — scatter-gather.
     *
     * Use when: write throughput is the constraint and scans are rare or
     * can tolerate multi-region fan-out.
     */
    public static byte[] saltedKey(String naturalKey, int regionCount) {
        // Ensure consistent positive modulus regardless of hashCode sign
        int salt = (naturalKey.hashCode() & 0x7FFFFFFF) % regionCount;
        // Zero-pad salt so lexicographic ordering of region ranges works correctly
        return Bytes.toBytes(String.format("%02d_%s", salt, naturalKey));
    }

    /**
     * Pattern 2: Reversed timestamp for time-series write distribution.
     *
     * Reversed timestamps decrease as real time increases, directing each
     * new write toward lower key space rather than always to the rightmost region.
     *
     * Bonus: 'latest N records for deviceId' scan reads contiguous rows
     * because recent data is now at lower keys, not scattered.
     *
     * Use when: ingesting time-series data (IoT, logs, events) and recent-data
     * reads are more important than historical range scans.
     */
    public static byte[] reversedTimestampKey(long timestampMillis, String deviceId) {
        long reversedTs = Long.MAX_VALUE - timestampMillis;
        // 16 hex chars = 8 bytes = full long range, zero-padded for correct lex sort
        return Bytes.toBytes(String.format("%016x_%s", reversedTs, deviceId));
    }

    /**
     * Pattern 3: Hash prefix for deterministic distribution.
     *
     * Same natural key always maps to the same region — lookups are fast
     * because you can reconstruct the hash from the natural key.
     * Trade-off: no range scan on natural key order; you lose locality.
     *
     * Use when: the workload is predominantly point Gets and range scans
     * on the natural key are not needed.
     */
    public static byte[] hashPrefixedKey(String naturalKey) {
        String prefix = md5HexPrefix(naturalKey, 4);  // 4-char prefix = 16^4 = 65536 buckets
        return Bytes.toBytes(String.format("%s_%s", prefix, naturalKey));
    }

    /**
     * Pattern 4: Composite key for multi-dimensional access patterns.
     *
     * Encodes multiple fields at fixed widths so lexicographic ordering
     * enables efficient range scans on the primary dimension while
     * filtering on secondary dimensions within the scan.
     *
     * Example: scan all events for US_EAST in reverse-chronological order
     * by setting startRow=US_EAST_<Long.MAX_VALUE> stopRow=US_EAST_<0>
     *
     * Use when: queries always filter on a primary dimension (region, tenant,
     * category) and sort or range-scan on a secondary dimension (time, score).
     */
    public static byte[] compositeKey(String regionCode, long timestampMillis, String userId) {
        // Fixed-width region code ensures correct lexicographic boundaries
        String paddedRegion  = String.format("%-6s", regionCode).replace(' ', '0');
        String reversedTs    = String.format("%016x", Long.MAX_VALUE - timestampMillis);
        return Bytes.toBytes(String.format("%s_%s_%s", paddedRegion, reversedTs, userId));
    }

    private static String md5HexPrefix(String input, int hexChars) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] hash = md.digest(Bytes.toBytes(input));
            StringBuilder sb = new StringBuilder();
            for (byte b : hash) {
                sb.append(String.format("%02x", b));
                if (sb.length() >= hexChars) break;
            }
            return sb.substring(0, hexChars);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException("MD5 unavailable — should never happen on a JVM", e);
        }
    }
}

package io.thecodeforge.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.RegionMetrics;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

import java.io.IOException;
import java.util.Collection;
import java.util.Map;

/**
 * Production compaction management — manual triggering, throughput throttling,
 * and configuration tuning to prevent compactions from overwhelming live traffic.
 *
 * The core rule: never let automatic major compactions run during peak hours.
 * For clusters serving live traffic, manual scheduling during off-peak windows
 * is the only safe approach at scale.
 */
public class CompactionManager implements AutoCloseable {

    private final Admin admin;
    private final Connection connection;

    public CompactionManager(String zkQuorum) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkQuorum);
        this.connection = ConnectionFactory.createConnection(conf);
        this.admin = connection.getAdmin();
    }

    /**
     * Trigger a major compaction on a specific table.
     * Call this during a scheduled maintenance window, not during peak hours.
     * Major compaction rewrites all HFiles — I/O impact is proportional to region size.
     */
    public void majorCompactTable(String tableName) throws IOException {
        System.out.printf("Triggering major compaction on table: %s%n", tableName);
        admin.majorCompact(TableName.valueOf(tableName));
        System.out.println("Major compaction submitted — runs asynchronously in the background");
    }

    /**
     * Trigger a minor compaction on a specific table.
     * Safe to run during business hours — merges a few HFiles with modest I/O.
     * Does NOT remove deleted cells or reclaim space from expired TTL data.
     */
    public void minorCompactTable(String tableName) throws IOException {
        admin.compact(TableName.valueOf(tableName));
    }

    /**
     * Check compaction queue depth per RegionServer.
     * A sustained queue > 5 indicates the RegionServer's disk I/O cannot keep up
     * with the incoming flush rate — time to tune or add capacity.
     */
    public void printCompactionQueueDepth() throws IOException {
        Map<ServerName, RegionMetrics> clusterMetrics =
            admin.getClusterMetrics().getLiveServerMetrics()
                 .entrySet().stream()
                 .collect(java.util.stream.Collectors.toMap(
                     Map.Entry::getKey,
                     e -> e.getValue().getRegionMetrics()
                               .values().stream().findFirst().orElse(null)
                 ));
        // In practice, iterate ServerMetrics for compactionQueueSize
        admin.getClusterMetrics().getLiveServerMetrics().forEach((server, metrics) -> {
            long compactionQueueSize = metrics.getCompactionQueueSize();
            System.out.printf("  %s  compaction queue: %d%n",
                server.getHostname(), compactionQueueSize);
        });
    }

    /**
     * Returns the recommended configuration for compaction tuning.
     *
     * Key decisions:
     *   - Disable automatic major compactions for clusters > 1TB
     *   - Throttle compaction throughput to leave headroom for live I/O
     *   - Increase minor compaction threshold to reduce frequency
     */
    public static Configuration getCompactionTunedConfig() {
        Configuration conf = HBaseConfiguration.create();

        // ── Minor compaction tuning ──────────────────────────────────────
        // Trigger minor compaction when 4+ HFiles exist per store (default: 3)
        // Higher threshold = less frequent compaction, more read amplification
        conf.setInt("hbase.hstore.compactionThreshold", 4);

        // Max HFiles merged in one minor compaction batch (default: 10)
        // Smaller batches = faster individual compactions, more total I/O over time
        conf.setInt("hbase.hstore.compaction.max", 6);

        // ── Major compaction tuning ──────────────────────────────────────
        // Disable automatic major compactions for large clusters
        // Set to 0 and schedule manually during off-peak windows
        conf.setLong("hbase.hregion.majorcompaction", 0L);  // 0 = disabled

        // If you do leave automatic major compaction enabled, add jitter
        // to prevent all regions from compacting simultaneously
        // conf.setLong("hbase.hregion.majorcompaction", 7 * 24 * 3600 * 1000L);
        // conf.setFloat("hbase.hregion.majorcompaction.jitter", 0.5f);

        // ── I/O throttling ───────────────────────────────────────────────
        // Limit compaction throughput to leave headroom for live reads/writes
        // 20MB/s per store is conservative; tune up if disk throughput allows
        conf.setLong("hbase.hstore.compaction.throughput.lower.bound",
                     20L * 1024 * 1024);  // 20 MB/s floor
        conf.setLong("hbase.hstore.compaction.throughput.higher.bound",
                     40L * 1024 * 1024);  // 40 MB/s ceiling

        return conf;
    }

    @Override
    public void close() throws IOException {
        admin.close();
        connection.close();
    }
}

package io.thecodeforge.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.regionserver.BloomType;

import java.io.IOException;

/**
 * Production read performance tuning for HBase.
 *
 * Covers:
 *   - BlockCache and MemStore memory budget (the 80% rule)
 *   - Bloom filter type selection per column family
 *   - Off-heap L2 bucket cache for larger working sets
 *   - Scan caching and result size limits
 *   - Column family compression
 *
 * The 80% rule is non-negotiable:
 *   hfile.block.cache.size + hbase.regionserver.global.memstore.size <= 0.80
 */
public class ReadTuningConfig {

    /**
     * Returns the recommended RegionServer configuration for a 32GB heap node.
     *
     * Memory budget:
     *   BlockCache : 40% of 32GB = 12.8GB  (on-heap L1)
     *   MemStore   : 35% of 32GB = 11.2GB
     *   Total      : 75%  (leaves 25% for GC, RPC handlers, JVM internals)
     *
     * With 8GB off-heap L2 bucket cache, effective cache = 12.8 + 8 = 20.8GB
     * without increasing GC pressure.
     */
    public static Configuration getReadOptimizedConfig() {
        Configuration conf = HBaseConfiguration.create();

        // ── On-heap BlockCache (L1) ──────────────────────────────────────
        // 40% of heap — primary cache for hot data
        conf.setFloat("hfile.block.cache.size", 0.40f);

        // ── MemStore ─────────────────────────────────────────────────────
        // 35% of heap — write buffer across all regions on this server
        // Triggers a global flush if total MemStore usage exceeds this
        conf.setFloat("hbase.regionserver.global.memstore.size", 0.35f);

        // Verify: 0.40 + 0.35 = 0.75 <= 0.80 — within the safe budget

        // ── Off-heap L2 Bucket Cache ─────────────────────────────────────
        // Stores larger working sets without contributing to GC pressure
        // Data is serialised to native memory — slightly slower than L1 but
        // allows a much larger effective cache without increasing -Xmx
        conf.set("hbase.bucketcache.ioengine", "offheap");
        conf.setLong("hbase.bucketcache.size", 8192L);  // 8192 MB = 8GB off-heap

        // ── Scan optimisation ────────────────────────────────────────────
        // Rows fetched per RPC during a Scan — default of 1 is catastrophically slow
        // 100-500 is reasonable; tune based on average row size
        conf.setInt("hbase.client.scanner.caching", 100);

        // Max bytes returned per scanner RPC — prevents OOM on wide rows
        conf.setLong("hbase.client.scanner.max.result.size", 2L * 1024 * 1024);  // 2MB

        return conf;
    }

    /**
     * Creates a table with per-column-family bloom filter and compression settings.
     *
     * Bloom filter type is set per column family because different families
     * can have fundamentally different access patterns on the same table.
     */
    public static void createOptimisedTable(Connection connection,
                                             String tableName) throws IOException {
        try (Admin admin = connection.getAdmin()) {
            HTableDescriptor tableDescriptor =
                new HTableDescriptor(TableName.valueOf(tableName));

            // Column family for point-lookup-heavy data — use ROW bloom filter
            // ROW: probabilistic skip on row key only
            // Low memory cost: ~1.2 bytes per unique row key at 1% false positive
            HColumnDescriptor metadataFamily = new HColumnDescriptor("meta");
            metadataFamily.setBloomFilterType(BloomType.ROW);
            metadataFamily.setCompressionType(Compression.Algorithm.SNAPPY);
            metadataFamily.setMaxVersions(1);     // time-series: keep only latest
            metadataFamily.setTimeToLive(86400 * 90);  // 90-day TTL
            tableDescriptor.addFamily(metadataFamily);

            // Column family for wide rows with selective column reads — use ROWCOL
            // ROWCOL: probabilistic skip on row key + column qualifier
            // Higher memory cost: 3-5x more than ROW — only justified for wide rows
            HColumnDescriptor attributesFamily = new HColumnDescriptor("attrs");
            attributesFamily.setBloomFilterType(BloomType.ROWCOL);
            attributesFamily.setCompressionType(Compression.Algorithm.LZ4);
            attributesFamily.setMaxVersions(3);   // retain last 3 versions
            tableDescriptor.addFamily(attributesFamily);

            // Column family for scan-heavy analytical data — no bloom filter
            // Bloom filters provide no benefit for range scans — skip the memory cost
            HColumnDescriptor metricsFamily = new HColumnDescriptor("metrics");
            metricsFamily.setBloomFilterType(BloomType.NONE);
            metricsFamily.setCompressionType(Compression.Algorithm.GZ);  // higher ratio for cold data
            metricsFamily.setMaxVersions(1);
            tableDescriptor.addFamily(metricsFamily);

            admin.createTable(tableDescriptor);
        }
    }

    /**
     * Selects the appropriate bloom filter type based on column family access pattern.
     * Use this as a decision aide when designing a new column family schema.
     */
    public static BloomType selectBloomType(boolean isGetHeavy,
                                             boolean hasWideRows,
                                             boolean isScanHeavy) {
        if (isScanHeavy) {
            return BloomType.NONE;  // scans read all blocks — bloom filters are wasted memory
        }
        if (isGetHeavy && hasWideRows) {
            return BloomType.ROWCOL;  // skip HFiles missing the specific column — expensive but targeted
        }
        if (isGetHeavy) {
            return BloomType.ROW;  // skip HFiles missing the row — cheap and effective
        }
        return BloomType.NONE;  // default for mixed or unknown patterns
    }
}