我对ApacheKafka非常陌生，我正在尝试执行kafka分区，同时将字符串发送给kafka，
做的和这里完全一样

public class Producer {
    public static void main(String[] argv)throws Exception {

           String topicName = "test";

           //Configure the Producer
           Properties configProperties = new Properties();
           configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.4.226:9092");
           configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");
           configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
           configProperties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,CountryPartitioner.class.getCanonicalName());
           configProperties.put("partitions.0","USA");
           configProperties.put("partitions.1","India"); 

           org.apache.kafka.clients.producer.Producer producer = new KafkaProducer(configProperties);
               ProducerRecord<String, String> rec = new ProducerRecord<String, String>(topicName,"message111");
               producer.send(rec, new Callback() {
                   public void onCompletion(RecordMetadata metadata, Exception exception) {
                       System.out.println("Message sent to topic ->" + metadata.topic() +" stored at offset->" + metadata.offset());
                   }
               });
           producer.close();
       }

}

现在我遇到了这样一个例外

2018-07-03 17:52:12 ERROR RecordBatch:102 - Error executing user-provided callback on message for topic-partition test-2:
java.lang.NullPointerException
    at com.spnotes.kafka.partition.Producer$1.onCompletion(Producer.java:29)
    at org.apache.kafka.clients.producer.internals.RecordBatch.done(RecordBatch.java:99)
    at org.apache.kafka.clients.producer.internals.RecordBatch.maybeExpire(RecordBatch.java:136)
    at org.apache.kafka.clients.producer.internals.RecordAccumulator.abortExpiredBatches(RecordAccumulator.java:220)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:192)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:141)
    at java.lang.Thread.run(Thread.java:748)

我做错了什么？提前谢谢。