EUREKA-732 - Async entitlement steps status tracking (revised plan)

Revises the original EUREKA-732 plan. The goal is unchanged; the mechanism is deliberately leaner:
no new status, no new table, no correlation heuristics.

Goal

mgr-tenant-entitlements (MTE) currently reports a flow as finished once its stages have run — but several stages only publish a Kafka message and return: CapabilitiesModuleEventPublisher,
SystemUserModuleEventPublisher, ScheduledJobModuleEventPublisher (all extend AbstractModuleEventPublisher). The real work happens later in mod-roles-keycloak,
mod-users-keycloak and mod-scheduler, and MTE never learns the outcome.

So finished means "messages dispatched", not "the tenant is usable". A consumer that fails leaves an entitlement that looks successful, and the failure only surfaces later as missing capabilities or
system users.

After this change, a flow reaches a terminal status only once the async work it triggered has reported
back.

Current state — what MTE stores today

Reference for the sections below. Nothing here changes except the one added column in section 1.

Flow persistence model

flow — one row per entitlement request. Entity FlowEntity + AbstractFlowEntity; PK flow_id.
Created IN_PROGRESS by FlowInitializer, finalized by FinishedFlowFinalizer / FailedFlowFinalizer / CancelledFlowFinalizer / CancellationFailedFlowFinalizer.

application_flow — one row per application in the request. Entity ApplicationFlowEntity; PK
application_flow_id (an @AttributeOverride of the inherited flow_id column), parent in flow_id. application_name / application_version are derived from application_id in @PrePersist. Created IN_PROGRESS by ApplicationFlowInitializer, finalized by the *ApplicationFlowFinalizer stages.

flow_stage — one row per executed stage. Entity FlowStageEntity, @IdClass(FlowStageKey), PK
(flow_id, stage). Written only by DatabaseLoggingStage (onStart inserts IN_PROGRESS;
onSuccess / onError / onCancel / onCancelError set the terminal status).

Details that matter for this design:

• flow_id on flow_stage is polymorphic. For module stages it is the application flow id
  (ModuleStageContext.getCurrentFlowId() → PARAM_APPLICATION_FLOW_ID); for common stages it is the top-level flow id. Hence the two-level cascade in section 4.

• There is no FK from flow_stage. The original constraint was dropped
  (dropForeignKeyConstraint fk_entitlement_stage_entitlement_flow); only idx_stage_flow_id and idx_stage_name remain. The parent reference is by convention.

• stage is a name, not an id. DatabaseLoggingStage.getStageName returns the class simple name;
  ModuleDatabaseLoggingStage overrides it to {moduleId}-{uncapitalizedClassName} (e.g.
  mod-foo-1.2.3-capabilitiesModuleEventPublisher). This is what makes an awaiting stage
  self-describing in the API.

• finished_at is @UpdateTimestamp on both flow entities and FlowStageEntity — Hibernate
  stamps it on insert as well as update, which is why it works as the publish-time anchor for
  trade-off 1.

• Enum values. entitlement_flow_status_type and entitlement_stage_status_type both hold
  QUEUED, IN_PROGRESS, CANCELLED, CANCELLATION_FAILED, FAILED, FINISHED (EntityExecutionStatus; the stage type was created without QUEUED, which is never used for stages). entitlement_flow_type holds ENTITLE, REVOKE, UPGRADE, STATE for flow
  (EntityFlowEntitlementType) and ENTITLE, REVOKE, UPGRADE for application_flow
  (EntityApplicationFlowEntitlementType).

• entitlement (PK (application_id, tenant_id), plus derived name/version) is the separate table written by EntitleApplicationFlowFinalizer — relevant to trade-off 4.

Events published today

Envelope. org.folio.integration.kafka.model.ResourceEvent<T> in folio-kafka-common:
id, type (CREATE / UPDATE / DELETE / DELETE_ALL), tenant (tenant name), resourceName,
new, old. Built by KafkaEventUtils.createEvent, which derives type from which payloads are
present and never sets id — the field this design reuses.

When tenantEntitlementKafkaProperties.isProducerTenantCollection() is set, the tenant segment is the
literal ALL instead of the tenant name (TOPIC_TENANT_COLLECTION_KEY). The Okapi-mode publishers
(AbstractEventPublisher) use the same topics with the tenant id as the message key.

Example capability event, trimmed:

Example system-user event payload — note name is the module name with the version stripped
(SystemUserEventProvider), which is why moduleId is a genuine addition in section 5:

Not a ResourceEvent: EntitlementEvent (type, moduleId, tenantName, tenantId), published
by EntitlementEventPublisher to {env}.entitlement — a non-tenant topic via getEnvTopicName, keyed {tenantName}_{moduleId}, consumed by sidecars and by mod-scheduler's entitlement-events listener. The new stage-result message is the second message of this shape: its own model, on an env-level topic, rather than a ResourceEvent payload.

Deviations from the original plan, and why

Scope

In scope: capability and system-user stages.

Out of scope for this ticket:

• mod-scheduler. Its event processing does not affect stage/flow status. ScheduledJobModuleEventPublisher keeps its current behaviour (onSuccess → finished). It can be revised later; see Follow-ups.

• A sweeper for stale awaiting stages. Needed, but a follow-up — see Follow-ups.

End-to-end sequence

The system-user path is identical with SystemUserModuleEventPublisher and mod-users-keycloak.

Contract

Topic

mgr-tenant-entitlements.stage-result, resolved with KafkaUtils.getEnvTopicName →
{env}.mgr-tenant-entitlements.stage-result.

A single topic, not split per tenant: the stage UUID is globally unique, so a topic per tenant
would only multiply topics without making lookups any easier, and the isProducerTenantCollection() per-tenant/per-collection duality of the outbound publishers does not apply.

The name keeps the mgr-tenant-entitlements. prefix used by the three existing topics
(.capability, .system-user, .scheduled-job) even though MTE is the consumer here: MTE owns the contract, and three producers / one consumer means naming by domain rather than by producer.

Message key and partitioning

The message key is the tenant name. One topic, but keyed so that tenants distribute across
partitions. This gives two properties that matter:

• Concurrency across tenants. With more than one partition and a listener concurrency above 1,
  several MTE consumer threads process confirmations for different tenants in parallel.

• Ordering within a tenant. All confirmations for a tenant land on the same partition and are
  consumed one at a time, so two stage results belonging to the same flow are never applied
  concurrently. That removes the interleaving that would otherwise make the flow-completion check ("are any stages still in_progress?") racy — two threads could each see the other's stage as pending and leave the flow in_progress forever.

Consequences for the implementation:

• The NewTopic bean must declare more than one partition for the concurrency to exist at all; keep
  it configurable, consistent with how the existing tenant topics are provisioned.

• Existing producers already have the tenant to hand — the outbound ResourceEvent carries tenant, so the confirming module echoes it into both the body and the key.

• Conditional (WHERE status = 'in_progress') updates in the listener stay, but their job narrows to
  idempotency — a duplicate or replayed delivery of the same confirmation — rather than
  guarding against concurrent writers to one flow. Same-tenant serialization is a property of the
  partitioning, not something the DB has to enforce.

• Cross-tenant flows do not exist (a flow belongs to one tenant_id), so tenant-keyed partitioning
  never splits one flow across partitions.

Message body

Where the contract lives

folio-integration-kafka / folio-kafka-common (in applications-poc-tools), next to ResourceEvent: the model, the two enums, the topic-name constant, and a small shared publisher component. The three downstream modules cannot depend on MTE, and the publisher is otherwise the same code three times.

Keeping the publisher shared also keeps the message key in one place: it sets the key from the
message's tenant rather than leaving each module to remember to do it. A module that keyed by stage id — or sent no key at all — would silently lose the per-tenant ordering the listener relies on, and nothing in the contract would catch it.

We will decide final code placement during implementation. Some code may remain inside modules

Carrying the stage id outbound

ResourceEvent.id is currently never populated — KafkaEventUtils.createEvent does not set it, and no consumer reads it. It carries the stage UUID at no cost and with no envelope change.

Set it on all three event types, including scheduled-job. mod-scheduler ignores it today, so when
it starts confirming there is no version skew.

Changes to Manager Tenant Entitlements

1. Stage UUID (additive)

flow_stage gets an id uuid column with a unique constraint. The existing (flow_id, stage)
primary key stays. This keeps FlowStageKey, DatabaseLoggingStage.setEntitlementStageStatus,
FlowStageService.getEntitlementStage and the public GET /application-flows/{id}/stages/{stageName} endpoint untouched, and the natural key remains a uniqueness guarantee.

Liquibase, new changeset under changes.v4.0.0/ (referenced from changelog-4.0.0.xml):

1. addColumn id uuid on flow_stage

2. backfill existing rows — UPDATE flow_stage SET id = gen_random_uuid() WHERE id IS NULL

3. addNotNullConstraint + addUniqueConstraint

FlowStageEntity gets the id field (not @Id). Expose it on the FlowStage DTO / schema so the
flow-details API can surface it — additive.

Generation and propagation. DatabaseLoggingStage.onStart generates the UUID when it inserts the row, and puts it into the stage context (context.put(ATTR_STAGE_ID, id)) so execute can attach it to the outbound ResourceEvent.

The id is generated exactly once per stage. DefaultStageExecutor.execute invokes onStart once per stage execution (folio-flow-engine), and MTE's retries are Spring Retry AOP interceptors inside execute (RetryConfiguration, @FolioModuleCallsRetryable, @KeycloakCallsRetryable) — they do not re-invoke onStart. Re-entitlement creates a new flow id and therefore new stage rows. So a stage id is stable for the life of the stage, and duplicate confirmations are always for the same id.

2. Keep the awaiting stage in_progress

DatabaseLoggingStage.onStart already sets IN_PROGRESS. The problem is onSuccess, which
unconditionally sets FINISHED as soon as execute returns.

• Add a protected hook to DatabaseLoggingStage — getSuccessStatus(C context), default FINISHED — and have onSuccess use it. Going through setEntitlementStageStatus keeps the existing retry-info flush and threadLocalModuleStageContext.clear() intact.

• Introduce a base class (or marker) for the async publishers, extended by
  CapabilitiesModuleEventPublisher and SystemUserModuleEventPublisher, which returns IN_PROGRESSonly when a message was actually published.

• AbstractModuleEventPublisher.execute must therefore record that it published. It frequently sends nothing — the isModuleUpdated / isModuleVersionChanged guards, and createEvent returning empty when both payloads are null. Those cases must still go FINISHED. A context attribute set at send time is enough.

Adding mod-scheduler later is then: extend the base class, and have that module confirm.

Resulting stage lifecycle:

3. Make the finalizers conditional

AbstractFlowFinalizer.execute unconditionally applies getFinalStatus(). Change it so that when
the final status would be FINISHED it first checks for stages still in_progress, and leaves the
flow in_progress if any exist. Guarding on FINISHED means the cancelled/failed finalizers are
unaffected, and entitle, upgrade and revoke application finalizers are all covered by one change.

Two levels, because for module stages FlowStageEntity.flowId is the application flow id
(ModuleStageContext.getCurrentFlowId() returns PARAM_APPLICATION_FLOW_ID):

• EntitleApplicationFlowFinalizer / UpgradeApplicationFlowFinalizer /
  RevokeApplicationFlowFinalizer → application_flow

• FinishedFlowFinalizer → flow: pending if any of its application_flow rows is in_progress, or any flow_stage row for the top-level flow id is in_progress

Needs a repository method along the lines of existsByFlowIdAndStatus(flowId, IN_PROGRESS).

EntitleApplicationFlowFinalizer keeps saving the Entitlement row as it does today — the entitlement is persisted before confirmations arrive. See Accepted trade-offs.

4. Stage-result listener

New Kafka listener on {env}.mgr-tenant-entitlements.stage-result, consumer group owned by MTE, concurrency above 1 so tenants are processed in parallel (see Message key and partitioning). The whole cascade for one confirmation runs in a single transaction, and stage/flow updates are conditional on the current status (WHERE status = 'in_progress') rather than read-modify-write, so a duplicate or replayed delivery is a no-op.