[
https://issues.apache.org/jira/browse/TIKA-5001?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132884#comment-18132884
]
Kristian Rickert commented on TIKA-5001:
----------------------------------------
*TIKA-5001: notes on steps 1-4, and where I'd take it*
this is the right direction. Couple of nits, then a longer-term thought below.
First - here's what I'd propose:
!proposed-arch.png|width=543,height=256!
h3. Nits
* Step 1: I'd actually delete the old store copies, not just warn. Otherwise
the secrets are still sitting on disk. Blocking runtime changes to
config-defined ids also feels like a v2 thing to me, since it changes
SaveFetcher/DeleteFetcher behavior. If we do v2, worth putting tika.proto under
buf lint/breaking while we're at it.
* Step 2: Lease needs to cover the whole job, not just the fork. Async emits
after the fork finishes, cancellation, shutdown. Watch
{{EmittingUnpackHandler}} here, it closes the shared emitter today.
* Step 3: Retire-everything on change is fine to start. The file store has a
concurrent writer problem too, two processes clobbering each other. Longer term
I'd rather have registration than the fingerprint anyway.
* Step 4: Yes, both. Kafka should wait (bounded) for the ack. JDBC needs the
whole transaction covered, reconnect included, not just the statement.
Can we keep the overall goals in 5001 and move the emitter fixes to linked
issues, as you suggested?
h3. Where I'd take it longer term
The biggest problem you'll get if you attempt this arch with millions of docs
is that the coordinator will need to track how busy the steps are or else it
will overwhelm the workers. This becomes 10x easier if you use to coordinator
as a broker instead of just a manager.
Workers can open one bidi stream, it can handshake with hello/register, then
pull work when they have capacity. The coordinator will track what work is
available by peeking into kafka. Otherwise the coordinator will need to track
state of downstream pipelines, it's far easier to have a contract that says "Is
work available?" and connect it with the demand for "I am a processing step X"
Registrations, leases, config revisions all belong in a DB. I started down the
Consul path on an earlier version of this, JDBC ended up way simpler. Avoid
zookeeper and raft for now, jdbc is easier.
I'd also make processors a first-class step. A processor is really just an
emitter that hands off to the next step. Run them as gRPC services rather than
forks (forks get you 100 zombies), whoever writes it owns it, language doesn't
matter since it is grpc.
To handle a pipeline you just need two leases, one per step and one for the
pipeline as a whole, with JDBC as the ledger and Kafka handling async/work
availability. And Tika should be able to emit more than one output from a run.
Rough proto attached to make it concrete: [^tika-coordinator-sample.proto]
I've built four of these over the last four years, so I can tell you the ton of
edge cases you'll run into as well.
> Improve runtime configuration in grpc
> -------------------------------------
>
> Key: TIKA-5001
> URL: https://issues.apache.org/jira/browse/TIKA-5001
> Project: Tika
> Issue Type: Task
> Reporter: Tim Allison
> Priority: Major
> Attachments: Simple tika pipes proposal.png, proposed-arch.png,
> tika-coordinator-sample-1.proto, tika-coordinator-sample.proto
>
>
> From my bot, based on ({*}[https://github.com/apache/tika/pull/3411]){*}
>
> {noformat}
> Summary. Runtime fetcher/emitter registration (gRPC
> SaveFetcher/DeleteFetcher) is only partly wired
> through tika-pipes: config-file components are copied into the config store
> with resolved
> secrets, updates and deletes never reach a fork that has already built the
> instance,
> and no fetcher or emitter is ever closed.
> Steps
> 1. Keep config-file components out of the store
> - Before: every process writes every config-file fetcher/emitter into the
> store at startup,
> ${env:...} resolved, so a file store holds secrets in plaintext and a
> shared Ignite store
> holds every node's config. A runtime save over a config-defined id is
> undone the next
> time a fork starts.
> - After: config-file components live in memory only; the store holds only
> runtime registrations
> as the caller sent them. A config-defined id can't be replaced or deleted
> at runtime;
> register a new id. Legacy store copies are ignored with a WARN. Upgrade
> note: delete store
> entries you didn't register at runtime.
> 2. Managers own component lifecycle (lease and close)
> - Before: one instance per id cached forever; nothing ever closes a fetcher
> or emitter, not on
> save/delete, not on shutdown (JDBCEmitter.close() never called).
> - After: acquire(id) returns a lease held for the whole task and released
> in its finally. A
> replaced/deleted instance is retired at once and closed when its last
> lease is released
> (immediately if idle). Everything is closed on graceful shutdown.
> Closeable components get
> called. API change: getFetcher/getEmitter return a lease. Docs: shared
> mode requires
> thread-safe fetchers and emitters.
> 3. Propagate store changes to running forks
> - Before: with file/ignite a fork picks up a new id on a cache miss, then
> caches it until
> restart, so updates and deletes are invisible. With memory, SaveFetcher
> succeeds and ListFetchers shows it, but every FetchAndParse fails
> FETCHER_NOT_FOUND.
> - After: per task, the fork compares the store fingerprint (file: content
> hash; ignite: counter)
> with the last seen; on change it retires all runtime components via step
> 2. Updates
> and deletes apply to the next task; a running task finishes with what it
> holds.
> Startup fails on allowComponentManagement: true + configStoreType:
> memory.
> 4. Shipped emitter bugs (own issues, linked)
> - KafkaEmitter: async send(), never flushed or acked; buffered records die
> with a recycled
> or killed fork. After: wait for the ack per emit; producer closed via
> step 2.
> - JDBCEmitter: concurrent emits share one PreparedStatement with no lock;
> shared mode can
> interleave parameters across rows. After: serialize the emit or use a
> connection per thread. {noformat}
>
> ||Store||Runtime registrations live in||Fork sees a new id||Fork sees an
> update or delete||Holds config-file copies (resolved secrets)||Survives gRPC
> restart||Broken today||
> |{{memory}} (default)|the gRPC JVM's heap only|No: {{SaveFetcher}} succeeds,
> {{ListFetchers}} shows it, every {{FetchAndParse}} through it fails
> {{FETCHER_NOT_FOUND}}|No|n/a (per process)|No|Yes, with
> {{allowComponentManagement}}; fine without it. Fixed by step 3 (startup fails
> on the combination)|
> |{{file}}|a JSON file every process reads and writes|Yes, on the fork's first
> cache miss|No, until the fork restarts|Yes, plaintext on disk|Yes|Yes:
> secrets on disk (step 1), stale updates/deletes (steps 2, 3), nothing closed
> (step 2)|
> |{{ignite}} (opt-in jars)|an Ignite cache shared across processes and
> nodes|Yes, on the fork's first cache miss|No, until the fork restarts|Yes,
> every node's config in
> one cache|Only with Ignite persistence|Yes: stale updates/deletes (steps 2,
> 3), nothing closed (step 2), cross-node config copies (step 1)|
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
