我正在尝试编译和运行一个简单的 kafka 作为示例的代码 Aapche 。编译时，即使添加了所有 lib 的文件 scala （我猜）。

Exception in thread "main" java.lang.NullPointerException
at scala.Predef$.Integer2int(Predef.scala:303)
at kafka.client.ClientUtils$$anonfun$parseBrokerList$1.apply(ClientUtils.scala:103)
at kafka.client.ClientUtils$$anonfun$parseBrokerList$1.apply(ClientUtils.scala:102)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:194)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:194)
at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:60)
at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:44)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:194)
at scala.collection.mutable.ArrayBuffer.map(ArrayBuffer.scala:44)
at kafka.client.ClientUtils$.parseBrokerList(ClientUtils.scala:102)
at kafka.producer.BrokerPartitionInfo.<init>(BrokerPartitionInfo.scala:32)
at kafka.producer.async.DefaultEventHandler.<init>(DefaultEventHandler.scala:41)
at kafka.producer.Producer.<init>(Producer.scala:60)
at kafka.javaapi.producer.Producer.<init>(Producer.scala:26)
at kafkaTest.TestProducer.main(TestProducer.java:23)

这是我的 program :

package kafkaTest;

import java.util.*;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class TestProducer {
public static void main(String[] args) {
    // long events = Long.parseLong(args[0]);

    long events = 10l;

    Random rnd = new Random();
    Properties props = new Properties();
    props.put("metadata.broker.list", "broker1:9092,broker2:9092 ");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
  ***![props.put("partitioner.class", "kafkaTest.SimplePartitioner");][1]***//this is line no 23
    props.put("request.required.acks", "1");
    ProducerConfig config = new ProducerConfig(props);
    Producer<String, String> producer = new Producer<String, String>(config);

      for (long nEvents = 0; nEvents < events; nEvents++) { long runtime =
      new Date().getTime(); String ip = "192.168.2.1" + rnd.nextInt(255);
      String msg = runtime + ",www.example.com," + ip; KeyedMessage<String,
      String> data = new KeyedMessage<String, String>("page_visits", ip,
      msg); producer.send(data); }

    producer.close();
}
}

附件是库文件的屏幕截图。请告诉我原因 error/exception .
编辑：这是 SimplePartitioner.java ```
package kafkaTest;

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

public class SimplePartitioner implements Partitioner {
public SimplePartitioner(VerifiableProperties props) {
}

public int partition(Object key, int a_numPartitions) {
int partition = 0;
String stringKey = (String) key;
int offset = stringKey.lastIndexOf('.');
if (offset > 0) {
partition = Integer.parseInt(stringKey.substring(offset + 1))
% a_numPartitions;
}
return partition;
}
}