[Feature] Support Fluss as ActionStateStore backend

[beryllw]

Linked issue: #627

Purpose of change

Uses Fluss log table + in-memory HashMap (similar to Kafka approach):

• Write: Sync append to Fluss log, then update cache
• Read: O(1) lookup from in-memory map
• Recovery: Incremental log scan from bucket offsets (recovery markers), last-write-wins
• Prune: Evict from cache only; physical cleanup via Fluss retention

Table schema: (state_key STRING, state_payload BYTES, agent_key STRING), distributed by agent_key.

• FlussActionStateStore — core implementation with divergence detection and data-loss validation
• ActionStateSerde — extracted backend-agnostic serializer (shared by Kafka & Fluss)
• ActionStateStore.BackendType.FLUSS — new enum value
• 9 new Fluss config options (bootstrap servers, database, table, buckets, SASL auth)
• Integration tests against embedded Fluss cluster (plain + SASL)
• Docs updated (configuration & deployment)

Tests

API

Documentation

• doc-needed
• doc-not-needed
• doc-included

[joeyutong]

Thanks for working on this. I’m also looking into Fluss as an ActionStateStore backend, so I’ve been comparing different designs here.

One question on the PK table approach: with synchronous upsert in put() and no-op rebuildState(), it seems backend upsert/lookup may sit on the hot path. For new seqNums, get() will usually need to hit Fluss, while the Kafka-style design keeps get() local by rebuilding post-checkpoint state into an in-memory cache.

Did you consider using a Log Table instead, plus local cache + recovery marker + changelog rebuild? My main concern is the latency tradeoff of putting PK-table lookup/upsert directly on the action execution path.

[joeyutong]

Thanks for your work on this.

[xintongsong]

LGTM, except for the code conflicts.

@Sxnan, as one of the contributor of the original Kafka action state store, could you also help check on this PR?

[Sxnan]

Thanks for the PR! I left a few comments belows.