kafka-go

kafka-go copied to clipboard

There is deployed kafka cluster of 3 nodes. The problem is on the consumer side, if the group coordinator has fallen, then when trying to synchronously commit, it does not switch to another broker and returns an error, and with an asynchronous commit it switches to another broker and continues to work. Why with a synchronous commit of messages it does not try to connect to another live broker?

Specif config reader:

func (c *Config) newReader(group *ConsumerGroup) *kafka.Reader {
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:          c.Brokers,
		Dialer:           c.getDialer(),
		GroupID:          group.GroupID,
		Topic:            group.Topic,
		MaxBytes:         MaxBytes,
		QueueCapacity:    QueueCapacity,
	})

	return r
}

Config of cluster in docker:

version: '2'
services:
  zookeeper-1:
    image: confluentinc/cp-zookeeper:latest
    hostname: zookeeper1
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 22181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zookeeper1:22888:23888;zookeeper2:32888:33888;zookeeper3:42888:43888
    # network_mode: host
    ports:
      - 22181:22181


  zookeeper-2:
    image: confluentinc/cp-zookeeper:latest
    hostname: zookeeper2
    environment:
      ZOOKEEPER_SERVER_ID: 2
      ZOOKEEPER_CLIENT_PORT: 32181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zookeeper1:22888:23888;zookeeper2:32888:33888;zookeeper3:42888:43888
    # network_mode: host
    ports:
      - 32181:32181

  zookeeper-3:
    image: confluentinc/cp-zookeeper:latest
    hostname: zookeeper3
    environment:
      ZOOKEEPER_SERVER_ID: 3
      ZOOKEEPER_CLIENT_PORT: 42181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
      ZOOKEEPER_SERVERS: zookeeper1:22888:23888;zookeeper2:32888:33888;zookeeper3:42888:43888
    # network_mode: host
    ports:
      - 42181:42181

  kafka-1:
    image: confluentinc/cp-kafka:latest
    # network_mode: host
    hostname: kafka1
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:22181,zookeeper2:32181,zookeeper3:42181
      KAFKA_LISTENERS: PLAINTEXT://kafka-1:19092,PLAINTEXT_HOST://0.0.0.0:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:19092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      # KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    ports:
      - 9092:9092

  kafka-2:
    image: confluentinc/cp-kafka:latest
    # network_mode: host
    hostname: kafka2
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:22181,zookeeper2:32181,zookeeper3:42181
      KAFKA_LISTENERS: PLAINTEXT://kafka-2:19093,PLAINTEXT_HOST://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:19093,PLAINTEXT_HOST://localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
    ports:
      - 9093:9093

  kafka-3:
    image: confluentinc/cp-kafka:latest
    # network_mode: host
    hostname: kafka3
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:22181,zookeeper2:32181,zookeeper3:42181
      KAFKA_LISTENERS: PLAINTEXT://kafka-3:19094,PLAINTEXT_HOST://0.0.0.0:9094
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:19094,PLAINTEXT_HOST://localhost:9094
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
    ports:
      - 9094:9094