class NotificationService extends GenericReactiveCrudService<NotificationEntity, String> {

    @Getter
    @Setter
    private BufferProperties buffer = new BufferProperties();

    private PersistenceBuffer<NotificationEntity> writer;

    public NotificationService() {
        buffer.setFilePath("./data/notification-buffer");
        buffer.setSize(1000);
        buffer.setTimeout(Duration.ofSeconds(1));
    }

    @PostConstruct
    public void init() {

        writer = new PersistenceBuffer<>(
            BufferSettings.create(buffer),
            NotificationEntity::new,
            flux -> this.save(flux).then(Reactors.ALWAYS_FALSE))
            .retryWhenError(err -> ErrorUtils.hasException(
                err,
                CannotCreateTransactionException.class,
                NonTransientDataAccessException.class,
                TimeoutException.class,
                IOException.class))
            .name("notification");

        writer.start();

    }

    @PreDestroy
    public void dispose() {
        writer.dispose();
    }

    @Subscribe("/notifications/**")
    public Mono<Void> subscribeNotifications(Notification notification) {
        writer.write(NotificationEntity.from(notification));
        return Mono.empty();
    }

    public Flux<NotificationEntity> findAndMarkRead(QueryParamEntity query) {<FILL_FUNCTION_BODY>}

}

        return this
            .query(query)
            .collectList()
            .filter(CollectionUtils::isNotEmpty)
            .flatMapMany(list -> this
                .createUpdate()
                .set(NotificationEntity::getState, NotificationState.read)
                .where()
                .in(NotificationEntity::getId, list
                    .stream()
                    .map(NotificationEntity::getId)
                    .collect(Collectors.toList()))
                .and(NotificationEntity::getState, NotificationState.unread)
                .execute()
                .thenMany(Flux.fromIterable(list))
                .doOnNext(e -> e.setState(NotificationState.read)));
    
<no_super_class>