Module 10 — High-Performance IPC: gRPC, Kafka & Event Streams

Implementing a Streaming gRPC Server in Node.js

import * as grpc from '@grpc/grpc-js';
import * as protoLoader from '@grpc/proto-loader';
 
const packageDef = protoLoader.loadSync('blockchain_indexer.proto', {
  keepCase: true,
  longs: String,
  enums: String,
  defaults: true,
  oneofs: true,
});
const proto = grpc.loadPackageDefinition(packageDef).blockchain;
 
// Server-streaming: push new blocks to all connected subscribers
function subscribeToBlocks(call) {
  const { startHeight } = call.request;
 
  // Register this subscriber
  const unsubscribe = blockEventBus.subscribe((block) => {
    if (block.height >= startHeight) {
      call.write({ block });  // push to this client
    }
  });
 
  call.on('cancelled', () => {
    unsubscribe();  // clean up when client disconnects
  });
 
  // Note: for server streaming, don't call call.end() until done
  // The stream stays open until the client cancels or the server calls end()
}
 
// Client-streaming: receive batches of transactions for indexing
async function ingestTransactions(call, callback) {
  const results = { ingested: 0, failed: 0 };
  const pendingWrites = [];  // track every in-flight write promise
 
  call.on('data', (transaction) => {
    // Do NOT await here — 'data' handlers can't block the stream from
    // delivering the next chunk. Instead, kick off the write and track its
    // promise so 'end' can wait for all of them before responding.
    const writePromise = writeTransaction(transaction)
      .then(() => { results.ingested++; })
      .catch(() => { results.failed++; });
    pendingWrites.push(writePromise);
  });
 
  call.on('end', async () => {
    // 'end' fires as soon as the client has sent its last chunk — it does NOT
    // mean the writes triggered by earlier 'data' events have resolved.
    // Without this, callback(null, results) can run before in-flight writes
    // finish, sending back a results object that's still being mutated,
    // or resolving before the last batch of transactions has actually landed.
    await Promise.allSettled(pendingWrites);
    callback(null, results);  // now every write has settled, one way or another
  });
}
 
const server = new grpc.Server();
server.addService(proto.BlockchainIndexer.service, {
  getTransaction: getTransactionHandler,
  subscribeToBlocks,
  ingestTransactions,
});
 
server.bindAsync('0.0.0.0:50051', grpc.ServerCredentials.createInsecure(), () => {
  server.start();
  console.log('gRPC server running on :50051');
});

gRPC Client with Retry and Deadline

const client = new proto.BlockchainIndexer(
  'indexer-service:50051',
  grpc.credentials.createInsecure(),
  {
    'grpc.keepalive_time_ms': 30_000,
    'grpc.keepalive_timeout_ms': 10_000,
    'grpc.keepalive_permit_without_calls': 1,
  }
);
 
// Unary call with deadline
function getTransaction(hash) {
  return new Promise((resolve, reject) => {
    const deadline = new Date(Date.now() + 5000);  // 5 second deadline
 
    client.getTransaction({ hash }, { deadline }, (err, response) => {
      if (err) reject(err);
      else resolve(response);
    });
  });
}
 
// Client-side streaming: send transactions in batches
function ingestBatch(transactions) {
  return new Promise((resolve, reject) => {
    const stream = client.ingestTransactions((err, result) => {
      if (err) reject(err);
      else resolve(result);
    });
 
    for (const tx of transactions) {
      stream.write(tx);
    }
    stream.end();
  });
}

gRPC vs REST: Throughput Comparison

gRPC is the correct choice for inter-service communication at > 10K calls/second or when bidirectional streaming is required.

---

Kafka: Distributed Event Ledger

Kafka is a distributed log. Unlike a message queue (which deletes messages after consumption), Kafka retains messages for a configurable retention period. Every consumer reads from its own offset in the log — Kafka does not push messages to consumers; consumers pull.

Core Concepts

Topic: a named log. transactions, blocks, payment-events.

Partition: a topic is split into partitions for parallelism. Each partition is an ordered, immutable log. Within a partition, messages are ordered by offset. Across partitions, there is no ordering guarantee.

Consumer group: a set of consumers that collectively read a topic. Kafka assigns partitions to consumers — each partition is consumed by exactly one consumer in the group. If you have 8 partitions and 8 consumers: each consumer reads one partition. If you have 8 partitions and 4 consumers: each consumer reads two partitions.

import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({
  clientId: 'blockchain-indexer',
  brokers: ['kafka1:9092', 'kafka2:9092', 'kafka3:9092'],
  ssl: true,
  sasl: { mechanism: 'scram-sha-256', username: 'indexer', password: process.env.KAFKA_PASS },
});

Producer: Batching and Compression

const producer = kafka.producer({
  // Batch settings for throughput
  maxInFlightRequests: 5,           // 5 simultaneous requests to broker
  idempotent: true,                  // exactly-once delivery (enables transactions)
  transactionTimeout: 60_000,
});
 
await producer.connect();
 
// High-throughput batch send
async function publishTransactionBatch(transactions) {
  await producer.sendBatch({
    topicMessages: [{
      topic: 'transactions',
      messages: transactions.map(tx => ({
        key: tx.hash,              // partition key — same sender always routes to same partition
        value: Buffer.from(serializeProtobuf(tx)),  // protobuf encoding
        timestamp: tx.timestamp.toString(),
      })),
    }],
    compression: CompressionTypes.LZ4,   // fast compression, good ratio
    acks: -1,                             // wait for all replicas (durability)
  });
}

acks: -1 alone is not a durability guarantee — it needs min.insync.replicas too. acks: -1 ("acks: all") tells the producer to wait for acknowledgment from every replica currently in the in-sync replica (ISR) set — but the ISR set can shrink. If two out of three replicas fall behind or go offline, the ISR can drop to just the leader, and acks: -1 is satisfied the instant that single leader writes the message — before it's actually replicated anywhere. A leader failure immediately afterward loses the message, even though the producer got a successful ack.

# broker-side topic config — set alongside acks: -1, not instead of it
min.insync.replicas=2   # with replication.factor=3, requires 2 of 3 replicas to ack
 
# What this actually buys you:
# - If the ISR shrinks below 2, the broker REJECTS the write outright
#   (producer gets NotEnoughReplicasException) instead of silently
#   accepting it durably on 1 replica.
# - acks: -1 + min.insync.replicas=2 is the combination that gives you
#   "this message survives losing any single broker." acks: -1 alone
#   only gives you "this message survives losing zero brokers, most
#   of the time, unless the ISR happened to shrink."

For a payment ledger's Kafka topics, set replication.factor=3 and min.insync.replicas=2 together — the pair is what "durable" actually means here, not acks: -1 in isolation.

Compression at 500K messages/second:

• LZ4: 2–3x compression, minimal CPU cost. Best for throughput.
• Snappy: 1.5–2x compression, very fast. Good default.
• GZIP: 3–5x compression, high CPU. Use for storage efficiency, not throughput.

Consumer Group: Partition Assignment and Rebalancing

const consumer = kafka.consumer({
  groupId: 'transaction-processor',
  sessionTimeout: 30_000,       // time before broker considers consumer dead
  heartbeatInterval: 3_000,     // send heartbeat every 3s
  maxBytesPerPartition: 1048576, // 1MB max per fetch per partition
});
 
await consumer.connect();
await consumer.subscribe({ topics: ['transactions'], fromBeginning: false });
 
// Rebalance listener — called when partition assignment changes
consumer.on(consumer.events.GROUP_JOIN, ({ payload }) => {
  console.log(`Joined group, assigned partitions: ${payload.memberAssignment}`);
});
 
await consumer.run({
  partitionsConsumedConcurrently: 4,  // process 4 partitions concurrently per consumer
  eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning }) => {
    for (const message of batch.messages) {
      if (!isRunning()) break;  // check for shutdown signal
 
      const tx = deserializeProtobuf(message.value);
      await processTransaction(tx);
 
      // Commit offset for this message
      resolveOffset(message.offset);
 
      // Heartbeat to prevent session timeout during long batches
      await heartbeat();
    }
  }
});

Consumer Lag: The Critical Operational Metric

Consumer lag = (latest offset in topic) - (consumer's current offset).

Lag = 0: consumer is caught up. Lag > 0: consumer is falling behind. Lag growing: consumer cannot keep up with producer rate.

// Monitor consumer lag from within Node.js
const admin = kafka.admin();
await admin.connect();
 
async function getConsumerLag() {
  const offsets = await admin.fetchOffsets({
    groupId: 'transaction-processor',
    topics: ['transactions'],
  });
 
  const topicOffsets = await admin.fetchTopicOffsets('transactions');
 
  let totalLag = 0;
  for (const partition of offsets[0].partitions) {
    const endOffset = topicOffsets.find(t => t.partition === partition.partition);
    const lag = parseInt(endOffset.offset) - parseInt(partition.offset);
    totalLag += lag;
 
    consumerLagGauge.labels(partition.partition.toString()).set(lag);
  }
 
  return totalLag;
}
 
setInterval(async () => {
  const lag = await getConsumerLag();
  if (lag > 500_000) {
    logger.error(`CRITICAL: consumer lag ${lag} — add more consumer instances`);
  }
}, 10_000);

Kafka Transactions: Exactly-Once Semantics Beyond the Idempotent Producer

idempotent: true on the producer (set earlier in this module) only solves duplicate delivery from retries — if a produce request times out and the client retries, the broker recognizes the duplicate sequence number and discards it, so the same message doesn't land twice in one partition. That's necessary for a payment ledger, but it's not sufficient on its own for a common pattern: read from topic A, process, write to topic B (or write to a DB and a topic together) without ever double-processing or dropping a message if the consumer crashes mid-write.

Kafka's transactional producer plus read_committed consumer isolation covers that gap — it lets a producer atomically commit a write to one or more topics together with the consumer offset that triggered it, so "I consumed this input and produced this output" becomes one atomic unit that either both happened or neither did.

// Transactional producer: consume-transform-produce as one atomic unit
const producer = kafka.producer({
  idempotent: true,
  transactionalId: 'settlement-processor-1',  // stable ID across restarts
  maxInFlightRequests: 5,
});
await producer.connect();
 
const consumer = kafka.consumer({ groupId: 'settlement-processor' });
await consumer.connect();
await consumer.subscribe({ topic: 'payment-events', fromBeginning: false });
 
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    const event = JSON.parse(message.value.toString());
 
    const transaction = await producer.transaction();
    try {
      // Produce the derived settlement event...
      await transaction.send({
        topic: 'settlements',
        messages: [{ key: event.accountId, value: JSON.stringify(computeSettlement(event)) }],
      });
 
      // ...and commit the consumer offset INSIDE the same transaction —
      // this is what makes "consume + produce" atomic
      await transaction.sendOffsets({
        consumerGroupId: 'settlement-processor',
        topics: [{ topic, partitions: [{ partition, offset: (Number(message.offset) + 1).toString() }] }],
      });
 
      await transaction.commit();
    } catch (err) {
      await transaction.abort();  // neither the offset nor the settlement event is persisted
      throw err;
    }
  },
});

// Downstream consumers of 'settlements' must read with read_committed isolation,
// otherwise they'll see uncommitted (and possibly later-aborted) messages
const downstreamConsumer = kafka.consumer({
  groupId: 'settlement-reader',
  readUncommitted: false,  // kafkajs default — only deliver committed transactional messages
});

Without read_committed, a downstream consumer using read_uncommitted isolation would see every message a transactional producer writes — including ones from transactions that later abort — defeating the entire point. read_committed (the correct default for a payment pipeline) buffers messages until the producing transaction commits or aborts, and only surfaces the committed ones, in order.

This is heavier than the idempotent producer alone — it costs an extra round trip per transaction and adds transaction.timeout.ms as a failure mode to monitor — so reach for it specifically at consume-transform-produce boundaries (like deriving settlement events from payment events) where "processed exactly once, never duplicated, never dropped" is a correctness requirement, not just a nice-to-have.

---

Event Sourcing for Payment Ledgers

Event sourcing stores every state change as an immutable event, instead of storing current state. The current state is derived by replaying all events.

Traditional approach:
  Table: accounts { id, balance, updated_at }
  UPDATE accounts SET balance = balance - 100 WHERE id = 'acc-1'
  → Current state only. History is gone.
 
Event sourcing:
  Table: account_events { id, account_id, event_type, amount, timestamp }
  INSERT INTO account_events VALUES ('evt-1', 'acc-1', 'DEBIT', 100, NOW())
  → Full history. Balance = SUM(credits) - SUM(debits).

Why event sourcing for payment ledgers:

1. Audit trail: every transaction is immutable. You can prove the exact sequence of operations that led to any balance.

2. Replay: if you discover a bug in your balance calculation logic, replay all events with the fixed logic to get correct current balances.

3. Temporal queries: what was account X's balance on March 15th? Replay events up to March 15th.

4. Kafka as the event store: Kafka's log retention makes it a natural event store.

// Payment ledger event sourcing with Kafka
 
// Write side: append events
async function debitAccount(accountId, amount, reference) {
  await producer.send({
    topic: 'payment-ledger',
    messages: [{
      key: accountId,  // same account always goes to same partition (ordering)
      value: JSON.stringify({
        eventType: 'ACCOUNT_DEBITED',
        eventId: crypto.randomUUID(),
        accountId,
        amount: amount.toString(),
        reference,
        timestamp: Date.now(),
      }),
    }],
    acks: -1,  // wait for all replicas before confirming
  });
}
 
// Read side: rebuild account state from events
async function getAccountBalance(accountId) {
  // Option 1: query the projection (maintained by consumer)
  const { rows } = await pool.query(
    'SELECT balance FROM account_projections WHERE account_id = $1',
    [accountId]
  );
  return rows[0]?.balance ?? 0n;
}
 
// Consumer: builds and maintains projection
consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value.toString());
 
    if (event.eventType === 'ACCOUNT_DEBITED') {
      await pool.query(
        `INSERT INTO account_projections (account_id, balance)
         VALUES ($1, 0 - $2)
         ON CONFLICT (account_id) DO UPDATE
         SET balance = account_projections.balance - $2`,
        [event.accountId, BigInt(event.amount)]
      );
    } else if (event.eventType === 'ACCOUNT_CREDITED') {
      await pool.query(
        `INSERT INTO account_projections (account_id, balance)
         VALUES ($1, $2)
         ON CONFLICT (account_id) DO UPDATE
         SET balance = account_projections.balance + $2`,
        [event.accountId, BigInt(event.amount)]
      );
    }
  }
});

Snapshots: Avoiding Full Replay

For accounts with millions of historical events, replaying from the beginning is impractical. Snapshots cache the balance at a point in time:

// Snapshot the account balance every 1,000 events
async function snapshotAccount(accountId, balance, lastEventOffset) {
  await pool.query(
    `INSERT INTO account_snapshots (account_id, balance, kafka_offset, created_at)
     VALUES ($1, $2, $3, NOW())
     ON CONFLICT (account_id) DO UPDATE
     SET balance = $2, kafka_offset = $3, created_at = NOW()`,
    [accountId, balance.toString(), lastEventOffset]
  );
}
 
// Replay from snapshot + subsequent events only
async function rebuildFromSnapshot(accountId) {
  const snapshot = await pool.query(
    'SELECT balance, kafka_offset FROM account_snapshots WHERE account_id = $1',
    [accountId]
  );
 
  if (!snapshot.rows[0]) return rebuildFromBeginning(accountId);
 
  const { balance, kafka_offset } = snapshot.rows[0];
  // Replay only events after the snapshot offset
  return replayFromOffset(accountId, kafka_offset, BigInt(balance));
}

---

RabbitMQ vs Kafka: When Each Is Correct

For a blockchain indexer: Kafka. You need replay capability (debug consumer bugs), high throughput (500K tx/sec), and consumer groups for parallel processing.

For payment notification emails: RabbitMQ. You need exactly-once delivery, message acknowledgement, dead letter queues, and retry logic. Kafka can do this but the configuration is more complex.

---

Production Incident: Kafka Partition Imbalance Causing Consumer Hotspot

Context: A payment event stream with 12 partitions, 12 consumer instances. Partition key was senderId. One major payment aggregator (a large e-commerce platform) sent 40% of all payment events. All events from this sender went to one partition.

What happened:

One consumer instance was processing 40% of total volume. Its consumer lag grew continuously. The other 11 instances were at 5% each. Rebalancing couldn't help — the problem was the partition key, not the number of consumers.

The fix — composite partition key:

// Before: senderId as partition key → all traffic from one sender → one partition
key: tx.senderId
 
// After: composite key hashes to distribute load
key: crypto.createHash('sha256')
  .update(`${tx.senderId}-${tx.timestamp % 1000}`)  // bucket by sender + time bucket
  .digest('hex')
  .slice(0, 16)
// Same sender's events are spread across ~1000 time buckets
// Still deterministic routing (same bucket always → same partition range)
// But no single partition receives 40% of traffic

Also added consumer lag alerting per partition (not just total lag) to detect hotspots early:

for (const { partition, lag } of partitionLags) {
  if (lag > avgLag * 3) {
    logger.warn(`Partition ${partition} hotspot: lag ${lag} vs avg ${avgLag}`);
  }
}

---

Summary

---

You now have the communication backbone. Module 11 covers how to keep the code clean as these systems grow — DDD applied to banking ledgers, dependency injection that lets you swap transports, and CQRS that separates the write model from the read model.

Next: Module 11 — Enterprise Architecture for State-Heavy Systems: DDD & Clean Architecture →

---