Skip to main content

Kafka Consumer Transport

Consume Telegram updates from a Kafka topic instead of polling or webhook.

Use Cases​

  • Decouple bot from Telegram API
  • Process updates through event streaming pipeline
  • Multiple bot instances consuming same topic
  • Real-time analytics on updates

Configuration​

application.yml:

telegram:
  bot:
    token: ${BOT_TOKEN}
    transport: KAFKA_CONSUMER
    kafka-consumer:
      topic: telegram-updates
      group-id: bot-group-1
      bootstrap-servers: localhost:9092

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: bot-group-1

Message Format​

Updates must be published as JSON in Kafka topic:

{
  "updateId": 123,
  "message": {
    "messageId": 456,
    "date": 1234567890,
    "chat": {"id": 789, "type": "private"},
    "from": {"id": 999, "firstName": "John"},
    "text": "Hello bot"
  }
}

Architecture​

Telegram API
    ↓
[Kafka Topic: telegram-updates]
    ↓
[Bot Instance 1] [Bot Instance 2] [Bot Instance 3]
    (all consuming same topic)

Example Setup​

docker-compose.yml:

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - "9092:9092"

  bot:
    build: .
    environment:
      TELEGRAM_BOT_TRANSPORT: KAFKA_CONSUMER
      SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092
      TELEGRAM_BOT_KAFKA_CONSUMER_TOPIC: telegram-updates
    depends_on:
      - kafka

Publishing Updates to Kafka​

Use messaging-kafka module to publish:

<dependency>
    <groupId>uz.osoncode.easygram</groupId>
    <artifactId>messaging-kafka</artifactId>
    <version>0.0.1</version>
</dependency>

application.yml:

telegram:
  bot:
    messaging:
      enabled: true
      type: KAFKA
      kafka:
        topic: telegram-updates
        bootstrap-servers: localhost:9092

See Broker Publishing for details.

Scaling​

Perfect for horizontal scaling:

# Bot instance 1
export BOT_CONSUMER_GROUP=bot-group-1
mvn spring-boot:run

# Bot instance 2 (same group)
export BOT_CONSUMER_GROUP=bot-group-1
mvn spring-boot:run

# Bot instance 3 (different group, processes all updates)
export BOT_CONSUMER_GROUP=bot-group-2
mvn spring-boot:run

Kafka automatically distributes partitions among instances in same group.

Next: Learn about RabbitMQ Consumer Transport.