Ресурсы companion на Java и Kotlin

Ресурс companion хранит объект, общий для джобов одного процесса JVM: например, клиент или справочные данные. Flow управляет его созданием, конфигурацией и очисткой. Java и Kotlin используют один API из tech.ytsaurus.flow.resource.

Сначала настройте JVM companion. Прокси на воркере, зависимости ресурсов и привязки компьютейшена описаны в общем справочнике.

Регистрация фабрики

Реализуйте FlowResource и до запуска companion зарегистрируйте в PipelineContext фабрику Supplier без аргументов:

  • registerResourceClass(String resourceClassName, Supplier<? extends FlowResource> factory) регистрирует явно заданное имя реализации.
  • registerResourceClass(Class<T> resourceClass, Supplier<? extends T> factory) получает имя реализации из класса.

Фабрика должна создавать новый экземпляр при каждом вызове. У разных идентификаторов ресурсов и у заменяющих экземпляров независимые жизненные циклы. Новый экземпляр может существовать одновременно с предыдущим, пока батч удерживает старый. Регистрация не вызывает фабрику. Повтор уже применённой инициализации не создаёт ещё один экземпляр. Пустое имя или повторная регистрация одного имени отклоняются.

При регистрации по классу используется непустое значение @FlowResourceClass. Если аннотации нет или её значение пусто, используется полное имя класса. Укажите это имя в parameters.companion_resource_class. Аннотация задаёт только имя: класс всё равно нужно зарегистрировать. При регистрации строкой используется переданное имя.

Имя реализации отличается от идентификатора в spec.resources и псевдонима компьютейшена. См. поддерживаемый тестовый companion и его спецификацию пайплайна.

Жизненный цикл и контекст

Метод

Назначение

load(ResourceContext context)

Инициализирует новый экземпляр по параметрам и зависимостям. Ожидаемые проверяемые исключения оборачивайте в ResourceLoadException, сохраняя причину.

reconfigure(ResourceContext context)

Применяет изменённые динамические параметры к работающему экземпляру. По умолчанию ничего не меняет.

unload()

Освобождает удерживаемые объекты и внешние ресурсы, в том числе после частично выполненного load. Метод нужно определить в каждой реализации; если освобождать нечего, реализация может быть пустой.

ResourceContext предоставляет resourceId(), parameters(), dynamicParameters() и dependencies(). Статические и динамические параметры имеют тип YTreeNode; при отсутствии динамических параметров передаётся пустой узел-словарь. Статические параметры включают companion_resource_class.

В словаре зависимостей ключом служит псевдоним зависимости, а если он не задан — её идентификатор. Вызов context.dependency(alias, ResourceType.class) проверяет ожидаемый тип. Отсутствие псевдонима или несовпадение типа вызывает IllegalStateException. Зависимости удерживаются всё время жизни зависимого экземпляра.

Чтение ресурса из пользовательского кода

В обработчике вызовите ctx.getResource(alias) или ctx.getResource(alias, ResourceType.class). Псевдоним берётся из required_resource_ids компьютейшена; если его нет, используйте идентификатор ресурса. Контекст обслуживающего companion вызывает IllegalStateException, если привязка отсутствует или тип не совпадает. Другие реализации RuntimeContext могут не поддерживать чтение ресурсов.

Получайте ресурс заново для каждого батча. Не кешируйте его в компьютейшене между батчами: следующему батчу может понадобиться другой экземпляр, а для сохранённого Flow уже может вызвать unload.

Конкурентный доступ и ошибки

Один экземпляр может одновременно обслуживать несколько джобов и потоков. Синхронизируйте доступ к изменяемым данным. До завершения reconfigure новые обращения за экземпляром отклоняются, но батчи и зависимые ресурсы, уже получившие его, могут продолжать работу. Удержание ресурса батчем откладывает unload; оно не синхронизирует изменения и не фиксирует конфигурацию.

Если load завершился ошибкой после создания экземпляра, Flow пытается выполнить очистку. После ошибки reconfigure экземпляр перестаёт быть доступен новым потребителям; последующая инициализация создаёт новый экземпляр и вызывает load. Уже выданные ссылки откладывают очистку до их освобождения. Исключение типа Exception из unload записывается в лог и подавляется; другие ошибки передаются вызывающему коду.

Воркер повторяет инициализацию. Предусмотрите повторную инициализацию и очистку после частичного создания ресурса. Управление жизненным циклом ресурса не делает внешние эффекты транзакционными.

Spring Boot и Kotlin

Для Spring Boot реализуйте ResourceProvider и зарегистрируйте провайдер как Spring bean. Метод getResourceClasses() возвращает Map<String, Supplier<? extends FlowResource>>. Каждый ключ — точное имя реализации для companion_resource_class; аннотация @FlowResourceClass для этого словаря не проверяется.

Провайдер может быть singleton, но его фабрики должны возвращать новые ресурсы. Методы жизненного цикла вызывает Flow, поэтому возвращать кешированный Spring singleton ресурса нельзя.

Kotlin использует те же интерфейсы, фабрики, аннотации и привязки в спецификации. Отдельный API ресурсов не нужен.

Исходный код и тесты

Предыдущая
Computation
Следующая
Флаг distribute