@galaxy-stack/orbit-microservices-kafka

Status: ✅ Full Implementation - Complete Kafka binary protocol với Bun native TCP

Mô tả

Kafka transport implementation cho Orbit microservices với consumer groups và partition assignment.

Tính năng

• Complete Kafka binary protocol encoder/decoder
• Consumer group protocol (JoinGroup/SyncGroup/Heartbeat)
• Range partition assignment strategy
• Partition leader discovery từ metadata
• Fetch và Produce operations
• Offset management với auto-commit
• Multi-broker connections với metadata refresh
• Produce error code checking
• Automatic reconnection
• Graceful shutdown

Cài đặt

bun add @galaxy-stack/orbit-microservices-kafka

Sử dụng

Kafka Server (Consumer)

import { BunFactory } from '@galaxy-stack/orbit-core';
import '@galaxy-stack/orbit-microservices-kafka';

const app = await BunFactory.createMicroservice(AppModule, {
  transport: 'KAFKA',
  options: {
    brokers: ['localhost:9092'],
    groupId: 'my-consumer-group',
    clientId: 'my-service',
    sessionTimeout: 30000,
    heartbeatInterval: 3000,
  },
});

await app.listen();
// [KafkaServer] Connected to Kafka cluster
// [KafkaServer] Joined group: my-consumer-group
// [KafkaServer] Listening...

Kafka Client (Producer)

import { MicroservicesModule } from '@galaxy-stack/orbit-microservices';
import '@galaxy-stack/orbit-microservices-kafka';

@Module({
  imports: [
    MicroservicesModule.register({
      name: 'KAFKA_SERVICE',
      transport: 'KAFKA',
      options: {
        brokers: ['localhost:9092'],
        clientId: 'my-producer',
        acks: -1,  // Wait for all replicas
      },
    }),
  ],
})
class ClientModule {}

Message Handlers

import { Controller, MessagePattern, EventPattern } from '@galaxy-stack/orbit-microservices';

@Controller()
class OrderController {
  @MessagePattern('orders.process')
  async processOrder(data: { orderId: number; items: any[] }) {
    return { success: true, orderId: data.orderId };
  }

  @EventPattern('orders.created')
  handleOrderCreated(data: { orderId: number }) {
    console.log('Order created:', data.orderId);
  }
}

Client Usage

@Injectable()
class OrderService {
  constructor(
    @Inject('KAFKA_SERVICE') private client: ClientProxy
  ) {}

  async processOrder(orderId: number, items: any[]) {
    return this.client.send('orders.process', { orderId, items });
  }

  notifyOrderCreated(orderId: number): void {
    this.client.emit('orders.created', { orderId });
  }
}

Options

KafkaServerOptions

interface KafkaServerOptions {
  brokers?: string[];             // Kafka brokers
  clientId?: string;              // Client identifier
  groupId?: string;               // Consumer group ID
  ssl?: boolean;                  // Enable SSL
  sasl?: {                        // SASL authentication
    mechanism: 'plain' | 'scram-sha-256' | 'scram-sha-512';
    username: string;
    password: string;
  };
  sessionTimeout?: number;        // Default: 30000ms
  rebalanceTimeout?: number;      // Default: 60000ms
  heartbeatInterval?: number;     // Default: 3000ms
  maxWaitTimeInMs?: number;       // Fetch wait time (default: 5000ms)
  minBytes?: number;              // Min fetch bytes (default: 1)
  maxBytes?: number;              // Max fetch bytes (default: 1MB)
  autoCommit?: boolean;           // Auto-commit offsets (default: true)
  autoCommitInterval?: number;    // Auto-commit interval (default: 5000ms)
}

KafkaProducerOptions

interface KafkaProducerOptions extends KafkaServerOptions {
  acks?: -1 | 0 | 1;              // Acknowledgment level
  timeout?: number;               // Request timeout (default: 30000ms)
  requestTimeout?: number;        // RPC timeout (default: 30000ms)
}

Kafka Protocol

Full implementation of Kafka binary protocol:

API Keys

| API | Key | Description | |-----|-----|-------------| | Produce | 0 | Send messages to topic | | Fetch | 1 | Fetch messages from topic | | Metadata | 3 | Get cluster metadata | | FindCoordinator | 10 | Find group coordinator | | JoinGroup | 11 | Join consumer group | | Heartbeat | 12 | Consumer heartbeat | | LeaveGroup | 13 | Leave consumer group | | SyncGroup | 14 | Sync group assignment | | ApiVersions | 18 | Get supported API versions |

Message Format

┌────────────────────────────────────────────────────────────────┐
│ Request/Response Header                                        │
├────────────┬─────────────┬──────────────┬─────────────────────┤
│ Size (4B)  │ API Key (2B)│ Version (2B) │ Correlation ID (4B) │
├────────────┴─────────────┴──────────────┴─────────────────────┤
│ Request/Response Body                                          │
└────────────────────────────────────────────────────────────────┘

Record Batch Format

┌─────────────────────────────────────────────────────────────────┐
│ First Offset (8B) │ Length (4B) │ Partition Leader Epoch (4B)  │
├─────────────────────────────────────────────────────────────────┤
│ Magic (1B) │ CRC (4B) │ Attributes (2B) │ Last Offset Delta (4B)│
├─────────────────────────────────────────────────────────────────┤
│ First Timestamp (8B) │ Max Timestamp (8B) │ Producer ID (8B)    │
├─────────────────────────────────────────────────────────────────┤
│ Producer Epoch (2B) │ First Sequence (4B) │ Records Count (4B)  │
├─────────────────────────────────────────────────────────────────┤
│ Records...                                                      │
└─────────────────────────────────────────────────────────────────┘

Topic Naming

| Type | Pattern | |------|---------| | Request | orbit-rpc-{pattern} | | Reply | orbit-reply-{clientId}-{timestamp} | | Event | orbit-event-{pattern} |

Consumer Groups

Kafka consumer groups with automatic rebalancing:

// Multiple instances with same groupId share partitions
options: {
  groupId: 'order-processors',
}

// Partition assignment:
// - Instance 1: partitions [0, 1]
// - Instance 2: partitions [2, 3]

Assignment Strategy

Currently supports Range assignment:

• Partitions are sorted and divided evenly among consumers
• Each consumer gets a contiguous range of partitions

KafkaConnection

Low-level Kafka connection với protocol support:

import { KafkaConnection } from '@galaxy-stack/orbit-microservices-kafka';

const conn = new KafkaConnection({
  brokers: ['localhost:9092'],
  clientId: 'my-client',
});

await conn.connect();

// Get metadata
const metadata = await conn.metadata(['my-topic']);

// Produce messages
const result = await conn.produce('my-topic', 0, [
  { key: Buffer.from('key'), value: Buffer.from('Hello!') }
]);

// Check produce errors
for (const topic of result.topics) {
  for (const partition of topic.partitions) {
    if (partition.errorCode !== 0) {
      console.error(`Produce error: ${partition.errorCode}`);
    }
  }
}

// Fetch messages
const fetch = await conn.fetch('my-topic', 0, 0n);
for (const record of fetch.records) {
  console.log(record.value.toString());
}

await conn.close();

Offset Management

// Manual offset commit
await conn.commitOffsets(groupId, generationId, memberId, [
  { topic: 'my-topic', partition: 0, offset: 100n }
]);

// Auto-commit (enabled by default)
options: {
  autoCommit: true,
  autoCommitInterval: 5000,
}

Error Codes

| Code | Name | Description | |------|------|-------------| | 0 | NONE | Success | | 1 | OFFSET_OUT_OF_RANGE | Offset out of range | | 3 | UNKNOWN_TOPIC | Unknown topic | | 6 | NOT_LEADER | Not partition leader | | 25 | REBALANCE_IN_PROGRESS | Rebalance in progress | | 27 | ILLEGAL_GENERATION | Illegal generation |

Error Handling

try {
  const result = await client.send('orders.validate', { orderId: 123 });
} catch (error) {
  console.error(error.message); // 'Order not found'
}

Reconnection

Automatic reconnection với group rejoin:

• Metadata refresh on broker failure
• Automatic group rejoin after disconnect
• Offset recovery from last committed

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                    Application                                   │
├─────────────────────────────────────────────────────────────────┤
│      KafkaServer           │         KafkaClient                │
├─────────────────────────────────────────────────────────────────┤
│              KafkaConnection (Broker Management)                 │
├─────────────────────────────────────────────────────────────────┤
│        KafkaEncoder / KafkaDecoder (Binary Protocol)            │
├─────────────────────────────────────────────────────────────────┤
│                   Bun.connect() (per broker)                    │
└─────────────────────────────────────────────────────────────────┘