Encountered error on one cluster complaining Kafka broker is 0, although 1 is up and running

ERROR admin.TopicCommand$: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 1 larger than available brokers: 0.

[donghua@cdh5 bin]$ ./kafka-topics --create --zookeeper cdh5:2181 --topic weblogs --replication-factor 1 --partitions 2

19/03/24 15:39:42 INFO zookeeper.ZooKeeper: Client environment:user.dir=/opt/cloudera/parcels/KAFKA-3.1.1-1.3.1.1.p0.2/bin
19/03/24 15:39:42 INFO zookeeper.ZooKeeper: Initiating client connection, connectString=cdh5:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@498d318c
19/03/24 15:39:42 INFO zkclient.ZkClient: Waiting for keeper state SyncConnected
19/03/24 15:39:42 INFO zookeeper.ClientCnxn: Opening socket connection to server cdh5.dbaglobe.com/192.168.31.25:2181. Will not attempt to authenticate using SASL (unknown error)
19/03/24 15:39:42 INFO zookeeper.ClientCnxn: Socket connection established, initiating session, client: /192.168.31.25:43992, server: cdh5.dbaglobe.com/192.168.31.25:2181
19/03/24 15:39:42 INFO zookeeper.ClientCnxn: Session establishment complete on server cdh5.dbaglobe.com/192.168.31.25:2181, sessionid = 0x169ae9e3aff0021, negotiated timeout = 30000
19/03/24 15:39:42 INFO zkclient.ZkClient: zookeeper state changed (SyncConnected)
Error while executing topic command : Replication factor: 1 larger than available brokers: 0.
19/03/24 15:39:42 ERROR admin.TopicCommand$: org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 1 larger than available brokers: 0.
19/03/24 15:39:42 INFO zkclient.ZkEventThread: Terminate ZkClient event thread.
19/03/24 15:39:42 INFO zookeeper.ZooKeeper: Session: 0x169ae9e3aff0021 closed
19/03/24 15:39:42 INFO zookeeper.ClientCnxn: EventThread shut down

Troubleshooting:

Check broker zookeeper path, which is /brokers/

[root@cdh5 log]# tail -f /var/log/kafka/kafka-broker-cdh5.dbaglobe.com.log 
2019-03-24 15:35:46,490 INFO kafka.utils.ZKCheckedEphemeral: Creating /brokers/ids/44 (is it secure? false)
2019-03-24 15:35:46,498 INFO kafka.utils.ZKCheckedEphemeral: Result of znode creation is: OK
2019-03-24 15:35:46,499 INFO kafka.utils.ZkUtils: Registered broker 44 at path /brokers/ids/44 with addresses: EndPoint(cdh5.dbaglobe.com,9092,ListenerName(PLAINTEXT),PLAINTEXT)

Check zookeeper.chroot configuration, which is /kafka". Change it “/” to fix the problem

[donghua@cdh5 bin]$ ./kafka-topics --create --zookeeper cdh5:2181 --topic weblogs --replication-factor 1 --partitions 2
19/03/24 15:49:53 INFO zkclient.ZkEventThread: Starting ZkClient event thread.
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:zookeeper.version=3.4.5-cdh5.14.2--1, built on 03/27/2018 20:39 GMT
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:host.name=cdh5.dbaglobe.com

19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:os.version=3.10.0-957.5.1.el7.x86_64
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:user.name=donghua
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:user.home=/home/donghua
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Client environment:user.dir=/opt/cloudera/parcels/KAFKA-3.1.1-1.3.1.1.p0.2/bin
19/03/24 15:49:53 INFO zookeeper.ZooKeeper: Initiating client connection, connectString=cdh5:2181 sessionTimeout=30000 watcher=org.I0Itec.zkclient.ZkClient@498d318c
19/03/24 15:49:53 INFO zkclient.ZkClient: Waiting for keeper state SyncConnected
19/03/24 15:49:53 INFO zookeeper.ClientCnxn: Opening socket connection to server cdh5.dbaglobe.com/192.168.31.25:2181. Will not attempt to authenticate using SASL (unknown error)
19/03/24 15:49:53 INFO zookeeper.ClientCnxn: Socket connection established, initiating session, client: /192.168.31.25:45328, server: cdh5.dbaglobe.com/192.168.31.25:2181
19/03/24 15:49:53 INFO zookeeper.ClientCnxn: Session establishment complete on server cdh5.dbaglobe.com/192.168.31.25:2181, sessionid = 0x169ae9e3aff003a, negotiated timeout = 30000
19/03/24 15:49:53 INFO zkclient.ZkClient: zookeeper state changed (SyncConnected)
19/03/24 15:49:54 INFO admin.AdminUtils$: Topic creation {"version":1,"partitions":{"1":[44],"0":[44]}}
Created topic "weblogs".
19/03/24 15:49:54 INFO zkclient.ZkEventThread: Terminate ZkClient event thread.
19/03/24 15:49:54 INFO zookeeper.ZooKeeper: Session: 0x169ae9e3aff003a closed
19/03/24 15:49:54 INFO zookeeper.ClientCnxn: EventThread shut down

How to delete Kafka topic message

Prepare the offset (message below it will be removed)

[donghua@hdp ~]$ /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --describe --topic kafka_hive_topic --zookeeper hdp:2181
Topic:kafka_hive_topic PartitionCount:1 ReplicationFactor:1 Configs:
Topic: kafka_hive_topic Partition: 0 Leader: 1001 Replicas: 1001 Isr: 1 001

parameter file:

[donghua@hdp ~]$ cat /tmp/delete_offset1.json 
{"partitions":                        
    [{"topic": "kafka_hive_topic", "partition": 0,   
     "offset": 12}],
   "version":1     

Execute kafka-delete-records.sh

[donghua@hdp ~]$ /usr/hdp/current/kafka-broker/bin/kafka-delete-records.sh --bootstrap-server hdp:6667 --offset-json-file /tmp/delete_offset1.json 
Executing records delete operation
Records delete operation completed:
partition: kafka_hive_topic-0 low_watermark: 12

Query Kafka topic directly using hive-kafka-storagehandler

1. Create kafka topic

[donghua@hdp ~]$ /usr/hdp/current/kafka-broker/bin/kafka-topics.sh --create --zookeeper hdp:2181 --replication-factor 1 --partitions 1 --topic kafka_hive_topic
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic "kafka_hive_topic".

2. Create hive table

[donghua@hdp ~]$ beeline -u "jdbc:hive2://hdp.dbaglobe.com:2181/demodb;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2" -n donghua -p x

Connecting to jdbc:hive2://hdp.dbaglobe.com:2181/demodb;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2

19/03/09 16:44:13 [main]: INFO jdbc.HiveConnection: Connected to hdp:10000

Connected to: Apache Hive (version 3.1.0.3.1.0.0-78)

Driver: Hive JDBC (version 3.1.0.3.1.0.0-78)

Transaction isolation: TRANSACTION_REPEATABLE_READ

Beeline version 3.1.0.3.1.0.0-78 by Apache Hive

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> 

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> CREATE EXTERNAL TABLE kafka_hive_table

. . . . . . . . . . . . . . . . . . . . . . >   (`Country Name` string , `Language` string,  `_id` struct<`$oid`:string>)

. . . . . . . . . . . . . . . . . . . . . . >   STORED BY 'org.apache.hadoop.hive.kafka.KafkaStorageHandler'

. . . . . . . . . . . . . . . . . . . . . . >   TBLPROPERTIES

. . . . . . . . . . . . . . . . . . . . . . >   ("kafka.topic" = "kafka_hive_topic", "kafka.bootstrap.servers"="hdp:6667");

No rows affected (4.747 seconds)

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> desc kafka_hive_table;

+---------------+----------------------+--------------------+

|   col_name    |      data_type       |      comment       |

+---------------+----------------------+--------------------+

| country name  | string               | from deserializer  |

| language      | string               | from deserializer  |

| _id           | struct<$oid:string>  | from deserializer  |

| __key         | binary               | from deserializer  |

| __partition   | int                  | from deserializer  |

| __offset      | bigint               | from deserializer  |

| __timestamp   | bigint               | from deserializer  |

+---------------+----------------------+--------------------+

7 rows selected (0.359 seconds)

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> !outputformat tsv2

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> !brief

verbose: off

createtab_stmt

CREATE EXTERNAL TABLE `kafka_hive_table`(

  `country name` string COMMENT 'from deserializer', 

  `language` string COMMENT 'from deserializer', 

  `_id` struct<$oid:string> COMMENT 'from deserializer', 

  `__key` binary COMMENT 'from deserializer', 

  `__partition` int COMMENT 'from deserializer', 

  `__offset` bigint COMMENT 'from deserializer', 

  `__timestamp` bigint COMMENT 'from deserializer')

ROW FORMAT SERDE 

  'org.apache.hadoop.hive.kafka.KafkaSerDe' 

STORED BY 

  'org.apache.hadoop.hive.kafka.KafkaStorageHandler' 

WITH SERDEPROPERTIES ( 

  'serialization.format'='1')

LOCATION

  'hdfs://hdp.dbaglobe.com:8020/warehouse/tablespace/external/hive/demodb.db/kafka_hive_table'

TBLPROPERTIES (

  'bucketing_version'='2', 

  'hive.kafka.max.retries'='6', 

  'hive.kafka.metadata.poll.timeout.ms'='30000', 

  'hive.kafka.optimistic.commit'='false', 

  'hive.kafka.poll.timeout.ms'='5000', 

  'kafka.bootstrap.servers'='hdp:6667', 

  'kafka.serde.class'='org.apache.hadoop.hive.serde2.JsonSerDe', 

  'kafka.topic'='kafka_hive_topic', 

  'kafka.write.semantic'='AT_LEAST_ONCE', 

  'transient_lastDdlTime'='1552121132')

27 rows selected (0.109 seconds)

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> 

3. Ingest some data into Kafka topic

[donghua@hdp ~]$ /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list hdp:6667 --topic kafka_hive_topic

>{"Country Name":"Afrika","Language":"af","_id":{"$oid":"55a0f1d420a4d760b5fbdbd6"},"ISO":0}

>{"Country Name":"Oseanië","Language":"af","_id":{"$oid":"55a0f1d420a4d760b5fbdbd7"},"ISO":0}

>^C

[donghua@hdp ~]$ /usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --topic kafka_hive_topic --bootstrap-server hdp:6667 --from-beginning

{"Country Name":"Afrika","Language":"af","_id":{"$oid":"55a0f1d420a4d760b5fbdbd6"},"ISO":0}

{"Country Name":"Oseanië","Language":"af","_id":{"$oid":"55a0f1d420a4d760b5fbdbd7"},"ISO":0}

^C

Processed a total of 2 messages

4. Query hive table

0: jdbc:hive2://hdp.dbaglobe.com:2181/demodb> select t.`Country Name` as Name, t.`Language` as lang, t.`__offset` from kafka_hive_table t;

INFO  : Compiling command(queryId=hive_20190309175638_f3a01692-28be-4683-bc66-32854615782c): select t.`Country Name` as Name, t.`Language` as lang, t.`__offset` from kafka_hive_table t

INFO  : Semantic Analysis Completed (retrial = false)

INFO  : Returning Hive schema: Schema(fieldSchemas:[FieldSchema(name:name, type:string, comment:null), FieldSchema(name:lang, type:string, comment:null), FieldSchema(name:t.__offset, type:bigint, comment:null)], properties:null)

INFO  : Completed compiling command(queryId=hive_20190309175638_f3a01692-28be-4683-bc66-32854615782c); Time taken: 0.289 seconds

INFO  : Executing command(queryId=hive_20190309175638_f3a01692-28be-4683-bc66-32854615782c): select t.`Country Name` as Name, t.`Language` as lang, t.`__offset` from kafka_hive_table t

INFO  : Completed executing command(queryId=hive_20190309175638_f3a01692-28be-4683-bc66-32854615782c); Time taken: 0.006 seconds

INFO  : OK

+----------+-------+-------------+

|   name   | lang  | t.__offset  |

+----------+-------+-------------+

| Afrika   | af    | 12          |

| Oseanië  | af    | 13          |

+----------+-------+-------------+

2 rows selected (0.379 seconds)

Additional Finding

if empty messages inside Kafka topic, the hive result could be wrong, as below:

offset 3 and 4 are empty string

Row 3 and 4 are incorrect, which repeat data from row 2.

ntp configuration to avoid KUDU ntptime error in lab/poc envronment

[root@cdh01 ~]# tail -n 5 /etc/ntp.conf
# Undisciplined Local Clock. This is a fake driver intended for backup
# and when no outside source of synchronized time is available.
server  127.127.1.0     # local clock
fudge   127.127.1.0 stratum 10

[root@cdh01 ~]# ntptime
ntp_gettime() returns code 0 (OK)
  time deb2b920.6f031890  Sat, May 26 2018  0:32:32.433, (.433641853),
  maximum error 133377 us, estimated error 527 us, TAI offset 0
ntp_adjtime() returns code 0 (OK)
  modes 0x0 (),
  offset 0.000 us, frequency -12.739 ppm, interval 1 s,
  maximum error 133377 us, estimated error 527 us,
  status 0x2000 (NANO),
  time constant 2, precision 0.001 us, tolerance 500 ppm,

=====================================

Alternative workaround:

Note: Use a Reliable clock in Linux.  Normally you need to install and configure NTP so Kudu has a reliable clock.  Without a reliable clock you may get errors starting and running Kudu that NTP would resolve. For development only, it is possible to avoid setting up and running ntpd with a reliable time server by running Kudu with the setting --use-hybrid-clock=false. However that setting has a serious effect on transactional consistency so it's not something we recommend for production or load testing.  Furthermore without a reliable clock provided by NTP, Kudu skips cleanup work it would normally do.  Your disk size will grow beyond what is normal and performance will be negatively impacted.5.For just this short lab only environment, because it is non-production, and we don’t care about performance in the lab, we will set the not recommended--use-hybrid-clock=false parameter.  Normally never forget to setup NTP for Kudu in production!

Click on the Kudu Service>Configuration. Look for the setting, “Kudu Service AdvancedConfiguration Snippet (Safety Valve) for gflagfile” and add the setting:

--use-hybrid-clock=false

Note: if you don’t set this clock setting correctly (or setup NTP), Kudu will fail to start intermittently and the full log file for the Kudu service will have errors indicating timesync problems such as:
Check failed:_s.ok() Bad status: Service unavailable: Cannot initialize clock: Error reading clock. Clock considered unsynchronized

Kafka hands-on with Docker

Reference URL: https://github.com/Landoop/fast-data-dev

docker run --rm --net=host -e ADV_HOST=192.168.1.86 landoop/fast-data-dev

# Start docker (change IP address to your Linux IP if external access needed, otherwise leave it to 127.0.0.1)
docker run --rm -it -p 2181:2181 -p 3030:3030 -p 8081:8081 -p 8082:8082 -p 8083:8083 -p 9092:9092 -e ADV_HOST=192.168.1.86 landoop/fast-data-dev

# Login the docker
docker run --rm -it --net=host landoop/fast-data-dev bash

root@fast-data-dev / $ kafka-topics --zookeeper 127.0.0.1:2181 --create --topic first_topic --partitions 3 --replication-factor 1
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic "first_topic".

root@fast-data-dev / $ kafka-topics --zookeeper 127.0.0.1:2181 --list
__consumer_offsets
_schemas
backblaze_smart
connect-configs
connect-offsets
connect-statuses
coyote-test-avro
coyote-test-binary
coyote-test-json
first_topic
logs_broker
nyc_yellow_taxi_trip_data
reddit_posts
sea_vessel_position_reports
telecom_italia_data
telecom_italia_grid

root@fast-data-dev / $ kafka-topics --zookeeper 127.0.0.1:2181 --create --topic second_topic --partitions 3 --replication-factor 1
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic "second_topic".
root@fast-data-dev / $ kafka-topics --zookeeper 127.0.0.1:2181 --delete --topic second_topic
Topic second_topic is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

root@fast-data-dev / $ kafka-topics --zookeeper 127.0.0.1:2181 --describe --topic first_topic
Topic:first_topic PartitionCount:3 ReplicationFactor:1 Configs:
Topic: first_topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Topic: first_topic Partition: 1 Leader: 0 Replicas: 0 Isr: 0
Topic: first_topic Partition: 2 Leader: 0 Replicas: 0 Isr: 0

root@fast-data-dev / $ kafka-console-producer --broker-list 127.0.0.1:9092 --topic first_topic
>first message
>second message
>^C


root@fast-data-dev / $ kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic

root@fast-data-dev / $ kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic --from-beginning

root@fast-data-dev / $ kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic --from-beginning --partition 0

# named consumer group commits its position in kafka hidden queue
root@fast-data-dev / $ kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic --consumer-property group.id=mygroup1 --from-beginning
second message
test
first message
test
test2
^CProcessed a total of 5 messages
# it will not read old message consumed before
root@fast-data-dev / $ kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic first_topic --consumer-property group.id=mygroup1 --from-beginning