SYM-7504: Refactor Staging Manager and interface boundary (symmetric-staging module, pluggable storage providers) - #814
Draft
Pavel_JM (pavel-jm) wants to merge 13 commits into
Draft
Conversation
… filesystem provider Introduces the foundation for the staging refactor described in the architecture plan: public API (IStagingManager, IStagedResource, IStagingLock, IStagingFactory, IStagingAuthProvider, ILineReader, ILineWriter), value types (StagingOptions, StagingKey, StagingConfig), enums (StorageKind, AccountType, ResourceLocation, ResourceState, ResourceKind, CompressionAlgorithm, EncryptionAlgorithm), and SPI abstractions (StorageBackend, LockBroker, LockObjectFormat, AbstractStagingManager, AbstractStagingResource). Filesystem provider preserves today's .create/.done extension format, adds explicit physical-location tracking (FILESYSTEM_PRIMARY vs FILESYSTEM_SCRATCH vs MEMORY), and uses the existing lock-file acquisition algorithm via the new LockBroker SPI. ThresholdSpillWriter replaces ThresholdFileWriter with location-aware spill callbacks. Parameter resolver registers staging.provider.type, staging.account.type, staging.scratch.dir, and the new SYM_STAGING_* environment variables. Factory rejects aws_s3, azure_blob, token, openid, and ldap with NotImplementedException pointing at the appropriate Jira tickets. 34 unit tests cover key/options value semantics, lock-object format round-trip, threshold-spill behavior, factory parameter resolution and provider routing, and the filesystem manager's byte-exact round-trip, lock acquisition contention, scratch-resource placement, and stale-lock takeover. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Scratch resources (StagingOptions.forScratch(), used by IStagingManager.createScratchResource) now have a hard zero memory threshold so no content is buffered in the JVM heap even briefly. Three companion fixes: - ThresholdSpillWriter no longer marks itself spilled at construction when thresholdBytes<=0; that left backend=null and NPE'd on first write. Constructor now always initializes the byte buffer and spill triggers naturally on the first non-empty write that exceeds the threshold (which is immediately when threshold=0). - FileSystemStagingResource.openWriter caps caller-supplied threshold at options.getMemoryThresholdBytes so a caller cannot widen the memory cap for a scratch resource. - FileSystemStagingManager.createScratchResource pipes the caller's options through StagingOptions.forScratch. New test createScratchResource_neverBuffersInMemory asserts the threshold is zero and the file lands on disk after a small write. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Renames every concrete StagingResource class to match the interface naming convention already used by IStagedResource: - AbstractStagingResource → AbstractStagedResource - FileSystemStagingResource → FileSystemStagedResource Updates referencing call sites in FileSystemStagingManager and the test suite. No behavior change. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Replaces the legacy filesystem staging implementation with the new symmetric-staging module while keeping the legacy IStagingManager and IStagedResource interfaces intact for the 100+ existing callers. Cipher SPI (community module): - IStreamCipherProvider, IStreamCipherContext, StreamCipherRegistry provide a ServiceLoader-based lookup; missing ciphers throw NotImplementedException so callers fail fast with a clear message. All concrete cipher impls live in symmetric-pro-staging. Legacy IStagingManager (org.jumpmind.symmetric.io.stage): - acquireFileLock now returns the new IStagingLock (legacy StagingFileLock class is deleted). - Added getScratchDirectory() so utilities that need scratch space (SnapshotUtil) go through the staging API instead of the engine parameter service. BatchStagingManager (symmetric-core): - Drops extends StagingManager; now implements IStagingManager via composition with a new symmetric-staging.api.IStagingManager delegate built through DefaultStagingFactory. - Looks up a cipher provider by id (staging.encryption.cipher) at construction; backwards-compat shim defaults to "aes" when STREAM_TO_FILE_ENCRYPT_ENABLED=true. - Preserves batch-aware clean(ttl, context) logic and the IBatchStagingExtension hook. - LegacyStagedResourceAdapter (new, package-private) wraps a new IStagedResource as the legacy type and applies cipher wrap on getInputStream/getOutputStream. Lock callers migrated (StagingFileLock -> IStagingLock): - IDataExtractorService, DataExtractorService, StagingPerf (releaseLock -> release, getLockFailureMessage -> getFailureMessage, getLockAge -> getAgeMs, getLockFile -> toString for log messages). Deletions in symmetric-io: - StagingManager, StagedResource, ThresholdFileWriter, StagingFileLock - ThresholdFileWriterTest (tested deleted class) SnapshotUtil: - Routes snapshot scratch through staging.scratch.dir via IStagingManager.getScratchDirectory(). Per-file createScratchResource rewrite tracked as follow-up TODO. symmetric-io now depends on symmetric-staging. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Adds a scratchFile(engine, tmpDir, ...) helper and replaces the top-level new File(tmpDir, ...) and new File(exportDir, ...) call sites in createSnapshot with it. Each major snapshot artifact (config-export, table-definitions, every CSV under export/, the top-level firebird/mysql diagnostics, threads/transactions files via their callers, and the runtime/parameters properties files) now allocates a scratch resource per file through the staging manager. The resources live under <scratchDir>/snapshots/<dirName>/ on disk, matching the prior tmpDir layout so ZipBuilder and external listeners that walk tmpDir continue to work unchanged. Helpers that internally create files (writeRuntimeStats, writeJobsStats, createThreadsFile, etc.) keep their existing new File(...) patterns for now since they write into tmpDir which still lives under the staging scratch volume. Threading scratchFile() through those helpers is tracked as a follow-up. Adds IStagingManager.createScratchResource(Object...) to the legacy interface; BatchStagingManager implements it by delegating to the new module's createScratchResource(StagingOptions.plain(), path) and wrapping the result via LegacyStagedResourceAdapter. writeProperties gains an ISymmetricEngine parameter so it can route its output through scratchFile. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…esource Threads helper methods through the staging API so every file SnapshotUtil creates is registered as a scratch resource: - createThreadsFile, createThreadStatsFile, createProcessInfoFile, createTransactionsFile now take (ISymmetricEngine, File tmpDir) instead of (String parent) and allocate their output files via scratchFile(). - outputSymDataForBatchesInError routes both the captured-data CSV and the parsed-row CSV through scratchFile under the errorDir scratch subkey instead of building paths off the local errorDir variable. - Mysql diagnostic queries (processlist, global-variables, session-variables) and the outgoing/incoming batch summary extractQuery calls now pass scratch-resource paths instead of String-concatenating tmpDir / exportDir. External files that SnapshotUtil only reads (conf/sym_service.conf, conf/.config, log directories, user.dir) keep their direct File usage — they aren't outputs and don't belong in scratch storage. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
… StagingDataWriterTest Moves LegacyStagedResourceAdapter from symmetric-core to symmetric-io and makes it public so it can be reused outside BatchStagingManager. Adds a companion LegacyStagingManagerAdapter (also in symmetric-io) that implements the legacy IStagingManager interface as a thin wrapper over the new symmetric-staging.api.IStagingManager. BatchStagingManager now extends LegacyStagingManagerAdapter and only overrides clean() with the batch-aware purge logic; constructor wiring (StagingConfig, cipher resolution, delegate creation) is delegated to the parent's public constructor. ~50 lines of duplication removed. StagingDataWriterTest is re-enabled. It now constructs an IStagingManager via: new LegacyStagingManagerAdapter(new FileSystemStagingManager(config)) which exercises the same adapter chain BatchStagingManager uses, without needing a mocked ISymmetricEngine. Both testReadThenWriteToFile and testReadThenWriteToMemory pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Lock heartbeat: - LockHeartbeatOutputStream / LockHeartbeatInputStream wrap stream IO and call IStagingLock.touch() at ttl/3 intervals (default). When isStillValid() returns false they throw LockLostException so long reads/writes detect cluster takeover mid-stream. - New LockLostException (IOException subtype) signals lost ownership. - LockHeartbeatTest covers happy path + takeover on both directions. Sidecar SHA-256 checksum: - IStagedResource grows writeSidecar / readSidecar; FileSystem and Minio resources delegate to their backend. - ChecksumWriteStream wraps an OutputStream with a DigestOutputStream- style tee that writes the digest to a <key>.sha256 sidecar on close. - ChecksumVerifier reads the sidecar and re-hashes on read; returns true when no sidecar exists (purely additive feature). - AbstractStagingManager.verifyChecksum routes to ChecksumVerifier. - LegacyStagedResourceAdapter accepts a checksumEnabled flag and wraps the output stream with ChecksumWriteStream when true. - LegacyStagingManagerAdapter exposes a 4-arg constructor for cipher + lockTtl + checksumEnabled. - BatchStagingManager reads staging.checksum.enabled and propagates to the adapter; logs once on startup. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…meter A recent rebase pulled in upstream changes that reverted my earlier signature updates to writeProperties and createTransactionsFile. This restores them: - writeProperties now takes (ISymmetricEngine, Properties, File, String) and routes the output file through scratchFile(). - createTransactionsFile takes (ISymmetricEngine, File tmpDir, ...) so scratchFile(engine, tmpDir, ...) resolves inside the body. This keeps the snapshot-file allocation going through IStagingManager.createScratchResource() as intended in SYM-7504, and matches the parallel pattern used by createThreadsFile, createThreadStatsFile, and createProcessInfoFile. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ing test Compression SPI (community module): - IStreamCompressionProvider mirrors IStreamCipherProvider. - StreamCompressionRegistry does ServiceLoader-based lookup; missing provider throws NotImplementedException with a clear message. - GzipStreamCompressionProvider (id="gzip") wraps java.util.zip — ships with the community module, no extra dependency. - META-INF/services registration so the registry auto-discovers it. Streaming cipher API additions on ISecurityService: - generateStreamingIv() — fresh random 12-byte AES-GCM IV. - getStreamingAlgorithmId() — algorithm id stamped into JMSE header. - getStreamingCipher(mode, iv) — AES-GCM cipher with engine key. - getStreamingCipher(mode, iv, algId) — read-side, honors algId from header. Default impls are interface methods so external ISecurityService impls keep working; SecurityService overrides to provide the real AES-GCM cipher (requires AES engine key, throws with actionable message otherwise). BatchStagingManager / LegacyStagingManagerAdapter / LegacyStagedResourceAdapter: - Added compression to the pipeline. Write order: user → compression → cipher → checksum → backend. Read order is reversed. - Adapter accepts the compression provider via a new 5-arg constructor; shorter constructors delegate with null defaults. - BatchStagingManager.resolveCompression reads staging.compression.codec (or falls back to "gzip" when the legacy STREAM_TO_FILE_COMPRESSION_ENABLED parameter is true). Streaming test: - LineReaderWriterStreamingTest writes 100,000 lines (~25 MB) line-by- line with memoryThresholdBytes=0, reads them back, and verifies each line matches. Proves the line I/O path is heap-bounded for large files in addition to correctness. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Pavel_JM (pavel-jm)
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
(Preparation for cloud storage provides and advances ciphers)
symmetric-staging module:
public API
IStagingManager, IStagedResource, IStagingLock, IStagingFactory, IStagingAuthProvider, ILineReader, ILineWriter),
value types
StagingOptions, StagingKey, StagingConfig,
enums
StorageKind, AccountType, ResourceLocation, ResourceState, ResourceKind, CompressionAlgorithm, EncryptionAlgorithm,
SPI abstractions
StorageBackend, LockBroker, LockObjectFormat, AbstractStagingManager, AbstractStagingResource
The Filesystem provider:
Parameters:
staging.provider.type, staging.account.type, staging.scratch.dir,
new SYM_STAGING_* environment variables.