机器有两块网卡，kafka的host.name开始只绑定在了内部IP上，另一块对外网卡无法访问，把值设置为空的话会kafka监听端口在所有的网卡上绑定，但是在外网访问时，客户端又遇到了java.nio.channels.ClosedChannelException异常信息，server端用tcpdump分析的时候发现客户端有传递kafka所在机器的机器名过来。在client里断点跟踪一下发现是findLeader的时候返回的元信息是机器名而不是IP。客户端无法解析这个机器名所以出现了前面的异常。

在server.properties 里还有另一个参数是解决这个问题的， advertised.host.name参数用来配置返回的host.name值，把这个参数配置为外网IP地址即可。

这个参数默认没有启用，默认是返回的java.net.InetAddress.getCanonicalHostName的值，在我的mac上这个值并不等于hostname的值而是返回IP，但在linux上这个值就是hostname的值。

遇到一个业务上的情况，在压力很大的情况下，某个topic的消息出现了小部分的冗余，发送端采用async模式。不确定kafka的async模式发送消息是否也会重试，文档里提到async模式发送增加了丢消息的可能，不知道message.send.max.retries参数是否也会起作用，群里讨论了一下，晚上跟踪了一下代码发现也是有 retry 的。

在 kafka.producer.async.DefaultEventHandler的handle方法里

while (remainingRetries > 0 && outstandingProduceRequests.size > 0) {
  ......
}

看来不论同步还是异步发送，消息重复的可能都是存在的。

在kafka的代码里，会看到Purgatory(炼狱), Reaper(死神) 等字眼，可见作者的文艺范儿十足，除了是卡夫卡迷之外，还可能是个但丁迷。借用他比较文学的字眼做一次标题党，这篇blog用来记录使用kafka过程中遇到的一些问题。

1) 测试环境单机多实例时log4j的问题

在同一台机器上启用多个kafka实例的话，除了server.properties要各自设置，还需要注意一下它们的log4j配置。虽然多个实例使用同一个日志配置运行也没有问题，但对于排查问题很不方便。所以各个实例，也需要自己的log4j配置。

但是在对每个实例指定了各自的log4j.properties之后，发现里面的kafka.logs.dir变量不管怎么定义，都是用的默认的路径。开启log4j的-Dlog4j.debug发现properties是被正确加载的，但并没有使用properties里的kafka.logs.dir这个变量的值，有可能是被系统环境变量给覆盖了，因为log4j.properties里的变量值，优先用系统环境变量，环境变量没有定义才用properties里变量的值。grep了一下果然，是在”kafka-run-class.sh”脚本里也声明了这个变量：

KAFKA_LOG4J_OPTS="-Dkafka.logs.dir=$LOG_DIR $KAFKA_LOG4J_OPTS"

要对各个实例设置不同的log4j日志路径，需要修改掉这里。

2) 关闭实例时zkClient线程异常，无法退出的情况

kafka server停止时，触发各个模块/线程的shutdown行为：

1) 停止 SocketServer
2) 停止 RequestHandlerPool  线程池
3) 停止 ReplicaManager
4) 停止 zkClient 线程
ok, shutdown completed

在zkClient的停止过程，如果遇到zookeeper端异常(挂掉)的情况，有可能陷入尝试重连的死循环，日志里看到它会一直尝试连接zookeeper失败，无法停止(可能需要多次kill甚至kill -9)。这个问题偶然发生，并不是必然，还没有仔细分析，暂没精力去跟踪。

WARN Session 0x14b498f4bda0006 for server null, unexpected error, closing socket connection and attempting reconnect (org.apache.zookeeper.ClientCnxn)

java.net.ConnectException: Connection refused
    at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
    at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:716)
    at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1146)

3) 基于SimpleConsumer轮询时的问题

在采用基于SimpleConsumer的消费端实现时，当消息已经处理完最新的一条时（消费端能力大于生产端），轮询的策略需要注意一下。在测试环境下，我们遇到过一个情况是大量的轮询导致整个测试环境网络的流量异常，原因是该topic一直没有新消息，consumer端的轮询没有设置等待参数，也没有在client线程里判断进行一个短暂的sleep。几乎是以死循环的方式不断跟server端通讯，尽管每次的数据包很小，但只要有几个这样的消费端足以引起网络流量的异常。

跟巨石聊了一下，发现这里有个参数maxWait可以设置(默认是0)，当服务器端没有新的消息时判断是否阻塞直到maxWait。不过在尝试的过程中发现这个参数单独使用并不work，还需要对minBytes也设置(设置一个大于0的值)才行，参见KafkaApis的handleFetchRequest方法：

val dataRead = readMessageSets(fetchRequest)
val bytesReadable = dataRead.values.map(_.messages.sizeInBytes).sum

if(fetchRequest.maxWait <= 0 ||
   bytesReadable >= fetchRequest.minBytes ||
   fetchRequest.numPartitions <= 0) {
  ...
  requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(response)))
} else {
  ...
  fetchRequestPurgatory.watch(delayedFetch)
}

这两个参数同时设置的情况下才有效：

val fetchReq = new FetchRequestBuilder().clientId(getClientId)
            .addFetch(kafkaTopic, kafkaPartition, offset, fetchSize)
            .maxWait(5000) 
            .minBytes(1)
            .build()

//2015.4.22 补充
在最新的0.8.2.1 版本里，kafka-run-class.sh脚本里做了改进，判断LOG_DIR变量为空才设置默认值，不像以前那样是写死的，这样可以在执行这个脚本之前声明日志路径。