当应用程序在本地 Kafka 实例上运行时，主题分区显示为空答案

【问题标题】：Topic partition appears as null when application is run on local Kafka instance当应用程序在本地 Kafka 实例上运行时，主题分区显示为空
【发布时间】：2021-06-08 18:55:43
【问题描述】：

我有这个制作人：

public class ProducerDemo {

    public static void main(String[] args) {
        
        String inputFile = "C:\\Users\\path\\to\\binary\\file";
        String bootstrapServers = "127.0.0.1:9092";
        
        try (
                InputStream inputStream = new FileInputStream(inputFile);
                OutputStream outputStream = new FileOutputStream(outputFile);
            ) {
     
                long fileSize = new File(inputFile).length();
     
                byte[] allBytes = new byte[(int) fileSize];
     
                inputStream.read(allBytes);
        
        // create producer properties
        Properties properties = new Properties();
        
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
        properties.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.cme.pe.kms.EncryptionInterceptor");
        
        // create the producer
        KafkaProducer<String, byte[]> producer = new KafkaProducer<String, byte[]>(properties);
        
        // create a producer record
        ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>("foo", allBytes);
    
        // send asynchronously
        producer.send(record);
        
        // flush and close the producer
        //producer.flush();
        producer.close();
        
        } catch (IOException ex) {
            ex.printStackTrace();
        }

    }

我也有Kafka Intereptor这个方法：

public ProducerRecord<String, byte[]> onSend(final ProducerRecord<String, byte[]> record) {
    try {
        MasterKeyProvider<?> key_provider = keySet.keyProvider(record.topic());
        Toppar toppar = new Toppar(record.topic(), record.partition(), 0L, 0);
        inboundDecoders.get(record.topic()+record.partition()).decodeMsg(record.value(), toppar);
        byte[] buf = encrypt(key_provider, record.value());

        return new ProducerRecord<>(record.topic(), record.partition(), record.key(), buf);
    } catch (Exception e) {
        LOGGER.error("unable to encrypt message", e);
        return new ProducerRecord<>(record.topic(), record.partition(), record.key(), new byte[0]);
    }
}

当我运行我的应用程序时，我收到了这个错误：

java.lang.NullPointerException 在 com.my.app.MyInterceptor.onSend(MyInterceptor.java:44) …… 在 com.my.app.ProducerDemo.main(ProducerDemo.java:50)

第 44 行是Toppar toppar = new Toppar(record.topic(), record.partition(), 0L, 0);。我认为问题在于record.partition() 为空。我在本地将 Kafka 作为单个实例运行。我以为分区会是 0。我看到的功能是否正确，或者我该如何解决这个问题？

【问题讨论】：

你需要展示 Toppar 类。单独传递 null 作为参数不会引发 NPE

标签： java apache-kafka nullpointerexception kafka-producer-api kafka-topic

【解决方案1】：

您对 record.partition() 为 null 的理解确实是正确的。

您在创建生产者记录的过程中没有提供主题或分区。不提供分区还是可以理解的，但你应该提供记录需要发送到的主题。

Kafka 有自己的自定义逻辑来确定分区，以防您没有在 ProducerRecord 中提供分区（跳过详细信息以保持简短）。如果您作为一些 POC 执行此操作，您可以使用 partition=0 创建生产者记录，这将解决您的问题。这一切都归结为您试图在拦截器中实现的目标。

【讨论】：