我能够用java编写一个kafka示例应用程序。它有3个主题和酒吧/酒吧工作良好。但无法将这些主题分配到不同的分区。
我的消费者

public class Consumers extends Thread {
    private static final List<String> TOPIC_LIST = Arrays.asList("topic1", "topic2", "topic3");
    private static final List<TopicPartition> PARTITION_LIST = 
Arrays.asList(new TopicPartition(TOPIC_LIST.get(0), 1), new TopicPartition(TOPIC_LIST.get(1), 2));

    private void message() {
        Properties consumerProperties = KafkaProperties.getConsumerProperties();
        org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProperties);
        consumer.assign(PARTITION_LIST);
        Logger.debug("Kafka IP : " + consumerProperties.getProperty("bootstrap.servers"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    process(record.topic(), record.value());
                }
            }
        } catch (Exception e) {
            Logger.error("error while consuming : " + e.getMessage());
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }

    private void process(String topic, String value) {
        KafkaProcessor.process(topic, value);
    }

    @Override
    public void run() {
        message();
    }
}

我的自定义分区器

public class CustomPartitioner implements Partitioner {
private static Map<String, Integer> partitionMap;

@Override
public void configure(Map<String, ?> configs) {
    System.out.println("Inside CustomPartitioner.configure " + configs);
    partitionMap = new HashMap<>();
    for (Map.Entry<String, ?> entry : configs.entrySet()) {
        if (entry.getKey().startsWith("partitions.")) {
            String keyName = entry.getKey();
            String value = (String) entry.getValue();
            int partitionId = Integer.parseInt(keyName.substring(11));
            partitionMap.put(value, partitionId);
        }
    }
}

@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    List partitions = cluster.availablePartitionsForTopic(topic);
    String valueStr = (String) value;
    String name = ((String) value).split(":")[0];
    if (partitionMap.containsKey(name)) {
        //If the country is mapped to particular partition return it
        return partitionMap.get(name);
    } else {
        //If no country is mapped to particular partition distribute between remaining partitions
        int noOfPartitions = cluster.topics().size();
        return value.hashCode() % noOfPartitions + partitionMap.size();
    }
}

public void close() {
}}

我的制作人

public void producer(String topic, String message) {
    Producer<String, String> producer = new KafkaProducer<>(KafkaProperties.getProducerProperties());
    try {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, null, message);
        producer.send(producerRecord);
        producer.close();
    } catch (Exception e) {
        Logger.error("kafka message publish error: ", e);
    }
}

当我运行这个代码的时候。我收到了以下警告。

[warn] o.a.k.c.p.ProducerConfig - The configuration partitions.2 = partition2 was supplied but isn't a known config.
[warn] o.a.k.c.p.ProducerConfig - The configuration partitions.1 = partition1 was supplied but isn't a known config.

我的制片人财产如下：，

properties.put("bootstrap.servers", "127.0.0.1:9092");
 properties.put("acks", "all");
 properties.put("retries", 0);
 properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getCanonicalName());
 properties.put("partitions.1", "partition1");
 properties.put("partitions.2", "partition2");
 properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
 properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

使用这些属性和代码，我无法发送或接收消息。我该怎么修。