Не удалось декодировать тип json для ключа: имя_файла в потоке потока данных Spring Cloud

Question

Я использую Spring Cloud Data Flow для настройки потока, который читает файл CSV, преобразует его с помощью специального процессора и регистрирует его:

stream create --name testsourcecsv --definition "file --mode=lines --directory=D:/toto/ --file.filename-pattern=adresses-28.csv --maxMessages=1000 | csvToMap --spring.cloud.stream.bindings.output.content-type=application/json | log --spring.cloud.stream.bindings.input.content-type=application/json" --deploy

Приложения file и csvToMap работают нормально, но в приложении журнала я вижу такое исключение для каждой записи:

2019-12-03 11:32:46.500 ERROR 1328 --- [container-0-C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$5  : Could not decode json type: adresses-28.csv for key: file_name

com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'adresses': was expecting ('true', 'false' or 'null')
 at [Source: (byte[])"adresses-28.csv"; line: 1, column: 10]
    at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1804) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:703) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3532) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2627) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:832) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:729) ~[jackson-core-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4141) ~[jackson-databind-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4000) ~[jackson-databind-2.9.9.jar!/:2.9.9]
    at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3091) ~[jackson-databind-2.9.9.jar!/:2.9.9]
    at org.springframework.cloud.stream.binder.kafka.BinderHeaderMapper.lambda$toHeaders$1(BinderHeaderMapper.java:268) ~[spring-cloud-stream-binder-kafka-2.1.4.RELEASE.jar!/:2.1.4.RELEASE]
    at java.lang.Iterable.forEach(Iterable.java:75) ~[na:1.8.0_202]
    at org.springframework.cloud.stream.binder.kafka.BinderHeaderMapper.toHeaders(BinderHeaderMapper.java:251) ~[spring-cloud-stream-binder-kafka-2.1.4.RELEASE.jar!/:2.1.4.RELEASE]

Это исключение также возникает для заголовка file_relativePath. Я не понимаю, почему spring-kafka пытается читать их как JSON.

Кроме того, приемник журнала правильно регистрирует мои записи:

2019-12-03 11:32:46.516  INFO 1328 --- [container-0-C-1] log-sink                                 : {"code_postal":"28200","id_fantoir":"28211_0127","source_nom_voie":"inconnue","numero":"1","code_insee":28211,"lon":1.260462,"code_insee_ancienne_commune":"","nom_afnor":"RUE DU VIEUX MOULIN","nom_voie":"Rue du Vieux Moulin","nom_ld":"","libelle_acheminement":"LOGRON","source_position":"inconnue","nom_commune":"Logron","nom_ancienne_commune":"","x":570633.27,"y":6784246.2,"alias":"","id":"28211_0127_00001","rep":"","lat":48.145756}

Я регистрирую заголовки kafka для отладки в моем процессоре csvToMap, что дает мне:

2019-12-03 11:32:37.042  INFO 10788 --- [container-0-C-1] c.d.streams.processor.CsvToMapProcessor  : headers {sequenceNumber=152963, file_name=adresses-28.csv, sequenceSize=0, deliveryAttempt=1, kafka_timestampType=CREATE_TIME, file_originalFile=NonTrustedHeaderType [headerValue="D:\\toto\\adresses-28.csv", untrustedType=java.io.File], kafka_receivedMessageKey=null, kafka_receivedTopic=testsourcecsv.file, file_relativePath=adresses-28.csv, kafka_offset=430949, scst_nativeHeadersPresent=true, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@7c3e63db, correlationId=9547c02d-e617-d981-f9b5-8df231530f66, kafka_receivedPartitionId=0, contentType=text/plain, kafka_receivedTimestamp=1575299282558, kafka_groupId=testsourcecsv}

Поэтому я окончательно не понимаю, почему приемник журнала пытается декодировать заголовки file_name и file_relativePath.

Я создал локальную среду с помощью:

• Windows 7
• Spring CDF server v 2.2.1.REALEASE
• Spring Cloud Skipper v 2.1.2.РЕЛИЗ
• Spring CDF shell v 2.2.1.RELEASE
• Кафка 2.12-2.3.0

Мой процессор csvToMap определяется следующим образом:

    @Component
    public class CsvToMapProcessor {
        private static final Logger LOGGER = LoggerFactory.getLogger(CsvToMapProcessor.class);

        @Autowired
        @Qualifier("csvMapper")
        private ObjectReader csvMapper;

        @Autowired
        @Qualifier("jsonWriter")
        private ObjectWriter jsonWriter;

        @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
        public Map<String, Object> transform(String csvLine, @Headers Map<String, Object> headers) {
            try {
                LOGGER.info("headers {}", headers);
                Map<String, Object> map = csvMapper.readValue(csvLine);
                return map;
            } catch (JsonProcessingException e) {
                LOGGER.error("An error occurs while reading CSV line {} : {}", csvLine, e.getMessage());
                LOGGER.debug(e.getMessage(), e);
                return null;
            }
        }
    }

с этим родителем:

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

И эта весенняя облачная версия:

<spring-cloud.version>Hoxton.RELEASE</spring-cloud.version>

Что я делаю не так, чтобы возникла эта проблема?

Answer 1

Так:

• поставляется с версией Spring-Cloud Hoxton, версия spring-cloud-stream - 3.0.0.RELEASE:

[INFO] +- org.springframework.cloud:spring-cloud-starter-stream-kafka:jar:3.0.0.RELEASE:compile
[INFO] |  \- org.springframework.cloud:spring-cloud-stream-binder-kafka:jar:3.0.0.RELEASE:compile
[INFO] |     +- org.springframework.cloud:spring-cloud-stream-binder-kafka-core:jar:3.0.0.RELEASE:compile
[INFO] |     |  \- org.springframework.integration:spring-integration-kafka:jar:3.2.1.RELEASE:compile
[INFO] |     \- org.springframework.kafka:spring-kafka:jar:2.3.3.RELEASE:compile

• приложение-приемник журнала 2.1.2 использует поток весенних облаков версии 2.1.4.RELEASE:

[INFO] +- org.springframework.cloud:spring-cloud-starter-stream-kafka:jar:2.1.4.RELEASE:compile
[INFO] |  \- org.springframework.cloud:spring-cloud-stream-binder-kafka:jar:2.1.4.RELEASE:compile
[INFO] |     +- org.springframework.cloud:spring-cloud-stream-binder-kafka-core:jar:2.1.4.RELEASE:compile
[INFO] |     |  \- org.springframework.integration:spring-integration-kafka:jar:3.1.0.RELEASE:compile
[INFO] |     \- org.springframework.kafka:spring-kafka:jar:2.2.8.RELEASE:compile

Как говорится в документации Spring-kafka 2.3.3 DefaultKafkaHeaderMapper.setEncodeStrings:

Установите значение true, если потребитель исходящей записи использует Spring для Apache Kafka версии ниже 2.3.

Приложение для сбора журналов на самом деле использует spring-kafka v 2.2.8, поэтому я должен установить для него значение true, используя настраиваемый преобразователь заголовков:

    @Bean("kafkaBinderHeaderMapper")
    public KafkaHeaderMapper kafkaBinderHeaderMapper() {
        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
        mapper.setEncodeStrings(true);
        return mapper;
    }

Но если я это сделаю, приемник журнала ничего не регистрирует из-за невозможности понять заголовок contentType, закодированный DefaultKafkaHeaderMapper. И команда предоставляет BinderHeaderMapper для решения этой проблемы:

Пользовательский преобразователь заголовков для Apache Kafka. Это идентично DefaultKafkaHeaderMapper из Spring Kafka. Это предназначено для решения некоторых проблем взаимодействия между приложениями Spring Cloud Stream 3.0.x и 2.x, где типы MIME, переданные как обычный MimeType в заголовке, не десериализуются должным образом.

Поэтому мне нужно настроить пользовательский BinderHeaderMapper в моем приложении:

    @Bean("kafkaBinderHeaderMapper")
    public KafkaHeaderMapper kafkaBinderHeaderMapper() {
        BinderHeaderMapper mapper = new BinderHeaderMapper();
        mapper.setEncodeStrings(true);
        return mapper;
    }

И все отлично работает.

Answer 2

Похоже, что значение, которое вы установили в --file.filename-patternесть проблемы. Можете ли вы проверить, действительно ли вы передаете значение, которое соответствуетAntPathMatcher (свойство filename-pattern основано на этом сопоставлении путей)?

Что произойдет, если вы попробуете что-то вроде --file.filename-pattern=*.csv?