Пытаюсь реализовать Rest API с использованием Kafka. POST запрос обрабатывается быстро очень быстро и никаких нареканий к нему нет. GET запрос отрабатывает не совсем корректно и с заметной задержкой времени.
Кусок лога от GET запроса:
'''
2021-09-21 17:08:17.651 INFO 204 --- [6.1-9001-exec-5] c.e.v.controller.FrameController : FrameController: Получены данные из Kafka за последние 50 секунд: []
2021-09-21 17:08:22.192 INFO 204 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.KafkaConsumer : [Consumer clientId=consumer-group_id-1, groupId=group_id] Seeking to offset 5648 for partition camera-0
2021-09-21 17:08:22.196 INFO 204 --- [ntainer#0-0-C-1] o.a.kafka.clients.consumer.Consumer : Kafka: Получен фрейм: Frame{name='frame0001.jpg', total=2, timestamp='2020-08-29 20:21:00', autonumbers=[аа001а34, аа001а34], url='http://192.168.56.103:9000/frames/21-09-2021/frame0001.jpg'}, metainfo: 1632233288747, camera, 5648, 0
2021-09-21 17:08:22.196 INFO 204 --- [ntainer#0-0-C-1] o.a.kafka.clients.consumer.Consumer : Kafka: Получен фрейм: Frame{name='frame0001.jpg', total=2, timestamp='2020-08-29 20:21:00', autonumbers=[аа001а34, аа001а34], url='http://192.168.56.103:9000/frames/21-09-2021/frame0001.jpg'}, metainfo: 1632233289829, camera, 5649, 0
2021-09-21 17:08:22.196 INFO 204 --- [ntainer#0-0-C-1] o.a.kafka.clients.consumer.Consumer : Kafka: Получен фрейм: Frame{name='frame0001.jpg', total=2, timestamp='2020-08-29 20:21:00', autonumbers=[аа001а34, аа001а34], url='http://192.168.56.103:9000/frames/21-09-2021/frame0001.jpg'}, metainfo: 1632233290937, camera, 5650, 0
2021-09-21 17:08:22.196 INFO 204 --- [ntainer#0-0-C-1] o.a.kafka.clients.consumer.Consumer : Kafka: Получен фрейм: Frame{name='frame0001.jpg', total=2, timestamp='2020-08-29 20:21:00', autonumbers=[аа001а34, аа001а34], url='http://192.168.56.103:9000/frames/21-09-2021/frame0001.jpg'}, metainfo: 1632233292091, camera, 5651, 0
'''
То есть, после того как через GET-запрос поступило время, на которое необходимо передвинуть offset, Listener понимает об этом только через 5 секунд (в данном случае, обычно время варьируется и составляет от 2 до 5 секунд). Почему возникает данная задержка и есть ли возможность через GET-запрос получить данные из Kafka за время = текущее время - время от GET-запроса.
Буду очень благодарен, за консультацию! Заранее спасибо!
Часть кода, которая относится к Consumer и GET-запросу.
Исходный код:
Конфигурационный файл:
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${spring.kafka.consumer.bootstrap-servers}")
private String bootstrapServers;
@Value("${spring.kafka.consumer.group-id}")
private String groupId;
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
configProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
configProps.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
configProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
configProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, "com.example.videocamerarestapi.model.Frame");
return configProps;
}
@Bean
public ConsumerFactory<String, Frame> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Frame> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, Frame> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}
Контроллер:
@RestController
@RequestMapping(value = "/frames")
public class FrameController {
KafkaProducerService kafkaProducerService;
KafkaConsumerService kafkaConsumerService;
MinioService minioService;
@Autowired
public FrameController(KafkaProducerService kafkaProducerService, KafkaConsumerService kafkaConsumerService, MinioService minioService) {
this.kafkaProducerService = kafkaProducerService;
this.kafkaConsumerService = kafkaConsumerService;
this.minioService = minioService;
}
private static final Logger logger = LoggerFactory.getLogger(FrameController.class);
@PostMapping
public ResponseEntity<?> addFrame(@RequestBody RequestFrame requestFrame)
{
Frame frame = new Frame();
String name = requestFrame.getName();
int total = requestFrame.getTotal();
String timestamp = requestFrame.getTimestamp();
String[] autonumbers = requestFrame.getAutonumbers();
String base64 = requestFrame.getBase64();
frame.setName(name);
frame.setTotal(total);
frame.setTimestamp(timestamp);
frame.setAutonumbers(autonumbers);
minioService.uploadFile(name, base64);
String url = minioService.getUrl(name);
frame.setUrl(url);
kafkaProducerService.send("camera",frame);
logger.info("FrameController: Новый фрейм был добавлен в MinIO и данные по нему переданы в Kafka: {}", frame);
return new ResponseEntity<>(HttpStatus.CREATED);
}
@GetMapping
public ResponseEntity<List<Frame>> getFrame(@RequestParam int time) {
kafkaConsumerService.getDataInTime(time);
List<Frame> frames = kafkaConsumerService.getFrames();
logger.info("FrameController: Получены данные из Kafka за последние {} секунд: {}", time, frames);
return new ResponseEntity<>(frames, HttpStatus.OK);
}
}
Сервис:
@Service
public class KafkaConsumerService extends AbstractConsumerSeekAware {
private static final Logger logger = LoggerFactory.getLogger(Consumer.class);
private List<Frame> frames = new ArrayList<>();
@KafkaListener(topics = "camera")
public List<Frame> consume(Frame frame, ConsumerRecordMetadata meta) {
frames.add(frame);
logger.info(String.format("Kafka: Получен фрейм: %s, metainfo: %d, %s, %d, %d", frame, meta.timestamp(), meta.topic(), meta.offset(), meta.partition()));
return frames;
}
public void getDataInTime(long time) {
long timestamp = System.currentTimeMillis() - time * 1000;
seekToTimestamp(timestamp);
}
public List<Frame> getFrames() {
return frames;
}
}
Модели:
public class Frame {
private String name;
private int total;
private String timestamp;
private String[] autonumbers;
private String url;
public Frame() {
}
public Frame(String name, int total, String timestamp, String[] autonumbers, String url) {
this.name = name;
this.total = total;
this.timestamp = timestamp;
this.autonumbers = autonumbers;
this.url = url;
}
public Frame(String name) {
this.name = name;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getTotal() {
return total;
}
public void setTotal(int total) {
this.total = total;
}
public String getTimestamp() {
return timestamp;
}
public void setTimestamp(String timestamp) {
this.timestamp = timestamp;
}
public String[] getAutonumbers() {
return autonumbers;
}
public void setAutonumbers(String[] autonumbers) {
this.autonumbers = autonumbers;
}
public String getUrl() {
return url;
}
public void setUrl(String url) {
this.url = url;
}
@Override
public String toString() {
return "Frame{" +
"name='" + name + '\'' +
", total=" + total +
", timestamp='" + timestamp + '\'' +
", autonumbers=" + Arrays.toString(autonumbers) +
", url='" + url + '\'' +
'}';
}
}
Настройки:
server.address = 192.168.56.1
server.port = 9001
spring.kafka.consumer.bootstrap-servers = 192.168.56.103:9092
spring.kafka.consumer.group-id = group_id
spring.kafka.producer.bootstrap-servers = 192.168.56.103:9092
minio.bucket.name=frames
minio.access.name=minio
minio.access.secret=miniostorage
minio.url=http://192.168.56.103:9000
spring.servlet.multipart.enabled=true
spring.servlet.multipart.file-size-threshold=2KB
spring.servlet.multipart.max-file-size=5MB
spring.servlet.multipart.max-request-size=10MB
Настройки логгирования:
log4j.rootLogger=DEBUG, STDOUT, file
log4j.appender.STDOUT=org.apache.log4j.ConsoleAppender
log4j.appender.STDOUT.layout=org.apache.log4j.PatternLayout
log4j.appender.STDOUT.layout.ConversionPattern=%5p [%t] (%F:%L) - %m%n
log4j.appender.file=org.apache.log4j.RollingFileAppender
log4j.appender.file.File=mylogs.log
log4j.appender.file.MaxFileSize=10MB
log4j.appender.file.MaxBackupIndex=10
log4j.appender.file.layout=org.apache.log4j.PatternLayout
log4j.appender.file.layout.ConversionPattern=%d{dd-MM-yyyy HH:mm:ss} %-5p %c{1}:%L - %m%n