Kafka Connect-由于意外对象类型而没有存在转换器：java.lang.double

Question

我有一个Kafka流应用程序，该应用程序根据主题的值，唱歌窗口进行一些计算。我读到，持续数据的最佳实践是将值推向另一个主题，并使用Kafka Connect从主题获取数据并将其保存在数据库中。

我下载了Confluent/Kafka-Connect图像，并通过使用DockerFile安装Mongodb-Kafka-Connector进行了扩展。

FROM confluentinc/cp-kafka-connect:7.0.3
COPY target/components/packages/mongodb-kafka-connect-mongodb-1.7.0.zip /tmp/mongodb-kafka-connect-mongodb-1.7.0.zip

RUN confluent-hub install --no-prompt /tmp/mongodb-kafka-connect-mongodb-1.7.0.zip

我将带有配置的请求发送到Kafka Connect：

 curl -X PUT http://localhost:8083/connectors/sink-mongodb-users/config -H "Content-Type: application/json" -d ' {
    "connector.class":"com.mongodb.kafka.connect.MongoSinkConnector",
    "tasks.max":"1",
    "topics":"MOVINGAVG",
    "connection.uri":"mongodb://mongo:27017",
    "database":"Temperature",
    "collection":"MovingAverage",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "key.converter.schemas.enable":true,
    "value.converter":"org.apache.kafka.connect.converters.DoubleConverter",
    "value.converter.schemas.enable":true
    }'

Kafka Streams应用程序生成记录：

StreamsBuilder streamsBuilder = new StreamsBuilder();
        KStream<String, Double> kafkaStreams = streamsBuilder.stream(topic, Consumed.with(Serdes.String(), Serdes.Double()));
        Duration timeDifference = Duration.ofSeconds(30);

        KTable table = kafkaStreams.groupByKey()
                .windowedBy(SlidingWindows.ofTimeDifferenceWithNoGrace(timeDifference))
                .aggregate(
                        () -> generateTuple(logger), // initializer
                        (key, value, aggregate) -> tempAggregator(key, value,aggregate, logger))
                .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
                .mapValues((ValueMapper<AggregationClass, Object>) tuple2 -> tuple2.getAverage());
        table.toStream().peek((k,v) -> System.out.println("Value:" + v)).to(targetTopic, Produced.valueSerde(Serdes.Double()));
        KafkaStreams streams = new KafkaStreams(streamsBuilder.build(), properties);
        streams.cleanUp();
        streams.start();

消息发送给主题，Mongo Connector正在阅读它们，但是抛出了一个例外：

Caused by: org.apache.kafka.connect.errors.DataException: Could not convert value `2.1474836E7` into a BsonDocument.

at com.mongodb.kafka.connect.sink.converter.LazyBsonDocument.getUnwrapped(LazyBsonDocument.java:169)

at com.mongodb.kafka.connect.sink.converter.LazyBsonDocument.containsKey(LazyBsonDocument.java:83)

at com.mongodb.kafka.connect.sink.processor.DocumentIdAdder.shouldAppend(DocumentIdAdder.java:68)

at com.mongodb.kafka.connect.sink.processor.DocumentIdAdder.lambda$process$0(DocumentIdAdder.java:51)

at java.base/java.util.Optional.ifPresent(Optional.java:183)

at com.mongodb.kafka.connect.sink.processor.DocumentIdAdder.process(DocumentIdAdder.java:49)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.lambda$buildWriteModel$1(MongoProcessedSinkRecordData.java:90)

at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)

at java.base/java.util.Collections$UnmodifiableCollection.forEach(Collections.java:1085)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.lambda$buildWriteModel$2(MongoProcessedSinkRecordData.java:90)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.tryProcess(MongoProcessedSinkRecordData.java:105)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.buildWriteModel(MongoProcessedSinkRecordData.java:85)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.createWriteModel(MongoProcessedSinkRecordData.java:81)

at com.mongodb.kafka.connect.sink.MongoProcessedSinkRecordData.<init>(MongoProcessedSinkRecordData.java:51)

at com.mongodb.kafka.connect.sink.MongoSinkRecordProcessor.orderedGroupByTopicAndNamespace(MongoSinkRecordProcessor.java:45)

at com.mongodb.kafka.connect.sink.StartedMongoSinkTask.put(StartedMongoSinkTask.java:75)

at com.mongodb.kafka.connect.sink.MongoSinkTask.put(MongoSinkTask.java:90)

at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:601)

... 10 more

Caused by: org.apache.kafka.connect.errors.DataException: No converter present due to unexpected object type: java.lang.Double

at com.mongodb.kafka.connect.sink.converter.SinkConverter.getRecordConverter(SinkConverter.java:92)

at com.mongodb.kafka.connect.sink.converter.SinkConverter.lambda$convert$1(SinkConverter.java:60)

at com.mongodb.kafka.connect.sink.converter.LazyBsonDocument.getUnwrapped(LazyBsonDocument.java:166)

... 27 more

为什么会抛出这样的例外？