[BUG] Bootstrapping fails if resources are not found

[riferrei]

Describe the bug

While testing KoP's ability to form a cluster capable of mimicquing Kafka's cluster protocol, I spun two brokers with KoP-enabled and the following occurs with version 2.10.1.4:

kafka-1  | 2022-08-14T23:09:23,049+0000 [main] ERROR io.streamnative.pulsar.handlers.kop.KafkaProtocolHandler - Failed to create offset metadata
kafka-1  | java.lang.IllegalStateException: org.apache.pulsar.client.admin.PulsarAdminException$NotFoundException: Topic not exist

Similarly, I tried the version 2.9.1.1 and I got a different — but related problem:

kafka-1 | 2022-08-14T23:23:22,981+0000 [main] ERROR org.apache.pulsar.broker.PulsarService - Failed to start Pulsar service: org.apache.pulsar.metadata.api.MetadataStoreException$BadVersionException: org.apache.zookeeper.KeeperException$BadVersionException: KeeperErrorCode = BadVersion for /counters/producer-name

To Reproduce Steps to reproduce the behavior:

1. Configure two or more brokers with KoP enabled
2. Start them at the same time (it's OK to miss for a few seconds)
3. The first one to get up-and-running stays running
4. The one that comes in seconds, will throw the errors above.

Here is the Docker Compose configuration for two Pulsar brokers I used for this test:

  kafka-1:
    image: streamnative/sn-pulsar:2.10.1.4
    container_name: kafka-1
    hostname: kafka-1
    command: >
      bash -c "bin/apply-config-from-env.py conf/broker.conf &&
      exec bin/pulsar broker"
    environment:
      PULSAR_PREFIX_messagingProtocols: kafka
      PULSAR_PREFIX_allowAutoTopicCreationType: partitioned
      PULSAR_PREFIX_kafkaTransactionCoordinatorEnabled: "true"
      PULSAR_PREFIX_brokerDeleteInactiveTopicsEnabled: "false"
      PULSAR_PREFIX_kafkaListeners: PLAINTEXT://0.0.0.0:9092
      PULSAR_PREFIX_kafkaAdvertisedListeners: PLAINTEXT://kafka-1:9092
      PULSAR_PREFIX_brokerEntryMetadataInterceptors: org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor
      PULSAR_PREFIX_zookeeperServers: zookeeper:2181
      PULSAR_PREFIX_configurationStoreServers: zookeeper:2181
      PULSAR_PREFIX_clusterName: pulsar-cluster
      PULSAR_PREFIX_managedLedgerDefaultEnsembleSize: 1
      PULSAR_PREFIX_managedLedgerDefaultWriteQuorum: 1
      PULSAR_PREFIX_managedLedgerDefaultAckQuorum: 1
    ports:
      - 6650:6650
      - 8080:8080
      - 9092:9092

  kafka-2:
    image: streamnative/sn-pulsar:2.10.1.4
    container_name: kafka-2
    hostname: kafka-2
    command: >
      bash -c "bin/apply-config-from-env.py conf/broker.conf &&
      exec bin/pulsar broker"
    environment:
      PULSAR_PREFIX_messagingProtocols: kafka
      PULSAR_PREFIX_allowAutoTopicCreationType: partitioned
      PULSAR_PREFIX_kafkaTransactionCoordinatorEnabled: "true"
      PULSAR_PREFIX_brokerDeleteInactiveTopicsEnabled: "false"
      PULSAR_PREFIX_kafkaListeners: PLAINTEXT://0.0.0.0:9093
      PULSAR_PREFIX_kafkaAdvertisedListeners: PLAINTEXT://kafka-2:9093
      PULSAR_PREFIX_brokerEntryMetadataInterceptors: org.apache.pulsar.common.intercept.AppendIndexMetadataInterceptor
      PULSAR_PREFIX_zookeeperServers: zookeeper:2181
      PULSAR_PREFIX_configurationStoreServers: zookeeper:2181
      PULSAR_PREFIX_clusterName: pulsar-cluster
      PULSAR_PREFIX_managedLedgerDefaultEnsembleSize: 1
      PULSAR_PREFIX_managedLedgerDefaultWriteQuorum: 1
      PULSAR_PREFIX_managedLedgerDefaultAckQuorum: 1
    ports:
      - 6651:6650
      - 8081:8080
      - 9093:9093

More information about the complete setup can be found here.

Expected behavior If two or more brokers with KoP enabled are started, they need to form a cluster quorum and avoid having brokers stopping their bootstrap because resources were not found.

Screenshots N/A

Additional context This test was made using Docker images from StreamNative.