案例：使用kafka做消息系统缓存ELK日志

搭建完成以后，使用kafka消费者测试时出现一下情况

[root@filebeat opt]# /opt/kafka/bin/kafka-console-consumer.sh --zookeeper 192.168.10.1:2181,192.168.10.2:2181,192.168.10.7:2181 --topic message --from-beginning 
zookeeper is not a recognized option
Option                                   Description                            
------                                   -----------                            
--bootstrap-server <String: server to    REQUIRED: The server(s) to connect to. connect to>                                                                   
--consumer-property <String:             A mechanism to pass user-defined       consumer_prop>                           properties in the form key=value to  the consumer.                        
--consumer.config <String: config file>  Consumer config properties file. Note  that [consumer-property] takes       precedence over this config.         
--enable-systest-events                  Log lifecycle events of the consumer   in addition to logging consumed      messages. (This is specific for      system tests.)                       
--formatter <String: class>              The name of a class to use for         formatting kafka messages for        display. (default: kafka.tools.      DefaultMessageFormatter)             
--from-beginning                         If the consumer does not already have  an established offset to consume     from, start with the earliest        message present in the log rather    than the latest message.             
--group <String: consumer group id>      The consumer group id of the consumer. 
--help                                   Print usage information.               
--isolation-level <String>               Set to read_committed in order to      filter out transactional messages    which are not committed. Set to      read_uncommitted to read all         messages. (default: read_uncommitted)
--key-deserializer <String:                                                     deserializer for key>                                                         
--max-messages <Integer: num_messages>   The maximum number of messages to      consume before exiting. If not set,  consumption is continual.            
--offset <String: consume offset>        The offset id to consume from (a non-  negative number), or 'earliest'      which means from beginning, or       'latest' which means from end        (default: latest)                    
--partition <Integer: partition>         The partition to consume from.         Consumption starts from the end of   the partition unless '--offset' is   specified.                           
--property <String: prop>                The properties to initialize the       message formatter. Default           properties include:                  print.timestamp=true|false            print.key=true|false                  print.value=true|false                key.separator=<key.separator>         line.separator=<line.separator>       key.deserializer=<key.deserializer>   value.deserializer=<value.            deserializer>                        Users can also pass in customized      properties for their formatter; more specifically, users can pass in      properties keyed with 'key.          deserializer.' and 'value.           deserializer.' prefixes to configure their deserializers.                 
--skip-message-on-error                  If there is an error when processing a message, skip it instead of halt.    
--timeout-ms <Integer: timeout_ms>       If specified, exit if no message is    available for consumption for the    specified interval.                  
--topic <String: topic>                  The topic id to consume on.            
--value-deserializer <String:                                                   deserializer for values>                                                      
--version                                Display Kafka version.                 
--whitelist <String: whitelist>          Regular expression specifying          whitelist of topics to include for   consumption.  

有经验的人这时候一看就是报错信息就知道语法错误
查阅资料发现
是版本的问题，老版本支持上面的语法，新版本如下
/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 节点测试的ip:9092 --topic 发布主题 --from-beginning

[root@filebeat opt]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 192.168.10.7:9092 --topic message --from-beginninghello
hi
lalalala

至于新旧版本怎么区分的，我也不是很清楚，这里我选择的kafka版本是
kafka_2.12-2.5.0.tgz,只能用第二种方法测试，
至于你们的呢？
就都试试吧，看看支持的是哪个