У меня есть маршрут Camel, который читает файл из S3 и обрабатывает входной файл следующим образом:
- Разберите каждую строку в POJO (студент) с помощью Bindy
- Разделить вывод по телу ()
- Агрегировать по атрибуту тела (
.semester) и размеру пакета 2 - Вызвать службу сохранения для загрузки в БД заданными партиями
Проблема в том, что при размере пакета 2 и нечетном количестве записей всегда есть одна запись, которая не сохраняется.
Предоставляется код Kotlin, но он не должен сильно отличаться от эквивалентного кода Java (за исключением косой черты перед "\ $ { простое выражение} "или отсутствие точек с запятой в завершающих операторах.
Если я установил размер пакета равным 1, то каждая запись будет сохранена, в противном случае последняя запись никогда не будет сохранена.
Я несколько раз проверял документацию по message-processor, но это не так. Похоже, что это касается именно этого сценария.
Я также установил [completionTimeout | completionInterval] в дополнение к completionSize, но это не имеет никакого значения.
Кто-нибудь раньше сталкивался с этой проблемой?
val csvDataFormat = BindyCsvDataFormat(Student::class.java)
from("aws-s3://$student-12-bucket?amazonS3Client=#amazonS3&delay=5000")
.log("A new Student input file has been received in S3: '\${header.CamelAwsS3BucketName}/\${header.CamelAwsS3Key}'")
.to("direct:move-input-s3-object-to-in-progress")
.to("direct:process-s3-file")
.to("direct:move-input-s3-object-to-completed")
.end()
from("direct:process-s3-file")
.unmarshal(csvDataFormat)
.split(body())
.streaming()
.parallelProcessing()
.aggregate(simple("\${body.semester}"), GroupedBodyAggregationStrategy())
.completionSize(2)
.bean(persistenceService)
.end()
С входным CSV-файлом, включающим семь (7) записей, это результат сгенерированного вывода (с некоторыми добавленными журналами отладки):
WARN 19540 --- [student-12-move] c.a.s.s.internal.S3AbortableInputStream : Not all bytes were read from the S3ObjectInputStream, aborting HTTP connection. This is likely an error and may result in sub-optimal behavior. Request only the bytes you need via a ranged GET or drain the input stream after use. INFO 19540 --- [student-12-move] student-workflow-main : A new Student input file has been received in S3: 'student-12-bucket/inbox/foo.csv' INFO 19540 --- [student-12-move] move-input-s3-object-to-in-progress : Moving S3 file 'inbox/foo.csv' to 'in-progress' folder... INFO 19540 --- [student-12-move] student-workflow-main : Moved input S3 file 'in-progress/foo.csv' to 'in-progress' folder... INFO 19540 --- [student-12-move] pre-process-s3-file-records : Start saving to database... DEBUG 19540 --- [read #7 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=7, name=Student 7, semester=2nd, javaMarks=25) DEBUG 19540 --- [read #7 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=5, name=Student 5, semester=2nd, javaMarks=81) DEBUG 19540 --- [read #3 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=6, name=Student 6, semester=1st, javaMarks=15) DEBUG 19540 --- [read #3 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=2, name=Student 2, semester=1st, javaMarks=62) DEBUG 19540 --- [read #2 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=3, name=Student 3, semester=2nd, javaMarks=72) DEBUG 19540 --- [read #2 - Split] c.b.i.d.s.StudentPersistenceServiceImpl : Saving record to database: Student(id=1, name=Student 1, semester=2nd, javaMarks=87) INFO 19540 --- [student-12-move] device-group-workflow-main : End pre-processing S3 CSV file records... INFO 19540 --- [student-12-move] move-input-s3-object-to-completed : Moving S3 file 'in-progress/foo.csv' to 'completed' folder... INFO 19540 --- [student-12-move] device-group-workflow-main : Moved S3 file 'in-progress/foo.csv' to 'completed' folder...
