feat(executors): Add Java connector support to PyAirbyte #746
Conversation
- Add JavaExecutor with automatic JRE management and TAR extraction - Add use_java parameter (None/True/False/Path) for Java execution control - Add use_java_tar parameter (None/Path) for connector TAR file location - Implement fallback logic: use_java_tar implies use_java=True when set - Add comprehensive documentation and error handling - Create source-snowflake example demonstrating Java connector usage - Copy implementation from PR #719 with updated dual-parameter API Requested by: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
Original prompt from AJ Steers |
🤖 Devin AI EngineerI'll be helping with this pull request! Here's what you should know: ✅ I will automatically:
Note: I can only respond to comments from users who have write access to this repository. ⚙️ Control Options:
|
👋 Greetings, Airbyte Team Member!Here are some helpful tips and reminders for your convenience. Testing This PyAirbyte VersionYou can test this version of PyAirbyte using the following: # Run PyAirbyte CLI from this branch:
uvx --from 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1754946348-java-connector-support' pyairbyte --help
# Install PyAirbyte from this branch for development:
pip install 'git+https://github.com/airbytehq/PyAirbyte.git@devin/1754946348-java-connector-support'Helpful ResourcesPR Slash CommandsAirbyte Maintainers can execute the following slash commands on your PR:
Community SupportQuestions? Join the #pyairbyte channel in our Slack workspace. |
aaronsteers
changed the title
There was a problem hiding this comment.
Pull Request Overview
This PR implements Java connector support for PyAirbyte by adding a new JavaExecutor class with dual-parameter API control. The implementation provides automatic JRE management, connector TAR extraction, and intelligent fallback logic between Docker and Java execution modes.
- Adds
JavaExecutorclass with automatic JRE download from Azul API - Introduces dual-parameter API (
use_javaanduse_java_tar) for fine-grained control - Implements intelligent fallback logic (Docker when available, Java when not)
Reviewed Changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
examples/run_source_snowflake_java.py |
Complete example script demonstrating Java connector usage with real-world patterns |
airbyte/sources/util.py |
Adds use_java and use_java_tar parameters to get_source() API |
airbyte/_executors/util.py |
Updates executor factory with dual-parameter fallback logic and Java support |
airbyte/_executors/java.py |
New JavaExecutor class implementing automatic JRE management and connector execution |
airbyte/_executors/__init__.py |
Exports the new JavaExecutor class in the module's public API |
| for executable in bin_dir.iterdir(): | ||
| if ( | ||
| executable.is_file() and executable.stat().st_mode & 0o111 | ||
| ): # Check if executable |
There was a problem hiding this comment.
The directory iteration and executable permission check could be vulnerable if the tar file contains symlinks or special files. Consider adding validation to ensure only regular files are processed and verify the executable is within expected bounds.
| ): # Check if executable | |
| # Ensure executable is a regular file, not a symlink or special file, and is within connector_dir | |
| try: | |
| resolved_path = executable.resolve() | |
| except RuntimeError: | |
| continue # Skip broken symlinks | |
| if ( | |
| not executable.is_symlink() | |
| and stat.S_ISREG(executable.stat().st_mode) | |
| and executable.stat().st_mode & 0o111 | |
| and str(resolved_path).startswith(str(self.connector_dir.resolve())) | |
| ): |
Copilot uses AI. Check for mistakes.
| }, | ||
| ) from ex | ||
|
|
||
| def _extract_jre_with_strip_components(self, tar: tarfile.TarFile) -> None: |
There was a problem hiding this comment.
I don't know what this means or if it's needed. Please expand the docstring or delete the method if not needed
…nts method - Add detailed explanation of tar strip-components=1 equivalent behavior - Document why JRE tar files need directory structure handling - Address GitHub comment from @aaronsteers requesting docstring expansion Addresses: #746 (comment) Co-Authored-By: AJ Steers <[email protected]>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughAdds a new JavaExecutor with JRE and connector-tar management, exposes it via package exports, extends executor selection with use_java/use_java_tar, propagates these options through source/destination helpers, and adds an example script demonstrating running a Java connector. No existing executor implementations were mutated. Changes
Sequence Diagram(s)sequenceDiagram
actor User
participant SrcDst as get_source/get_destination
participant Factory as get_connector_executor
participant Exec as Executor (Java/Docker/Python)
User->>SrcDst: call(..., use_java, use_java_tar, ...)
SrcDst->>Factory: get_connector_executor(..., use_java, use_java_tar, ...)
alt use_java/use_java_tar provided or JAVA metadata + fallback
Factory-->>SrcDst: JavaExecutor
else
Factory-->>SrcDst: Docker/Venv/Path/Declarative Executor
end
SrcDst-->>User: Source/Destination bound to Executor
sequenceDiagram
actor Caller
participant JExec as JavaExecutor
participant FS as Filesystem/Cache
participant Azul as Azul API
participant Proc as Subprocess
Caller->>JExec: ensure_installation()
alt JRE missing and not provided
JExec->>Azul: query JRE tar (os/arch)
Azul-->>JExec: JRE tar URL
JExec->>FS: download + extract (strip top dir)
end
alt Connector tar provided
JExec->>FS: extract connector tar to ~/.airbyte/connectors/{name}
end
Caller->>JExec: execute(args, stdin?)
JExec->>Proc: spawn (JAVA_HOME/PATH set)
Proc-->>JExec: stdout lines
JExec-->>Caller: yield lines
Proc-->>JExec: exit code
JExec-->>Caller: raise on non-zero exit
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Would you like me to run a quick grep of call sites for the new ✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR/Issue comments)Type Other keywords and placeholders
CodeRabbit Configuration File (
|
|
Note Other AI code review bot(s) detectedCodeRabbit has detected other AI code review bot(s) in this pull request and will avoid duplicating their findings in the review comments. This may lead to a less comprehensive review. 📝 WalkthroughWalkthroughAdds Java connector execution support. Introduces JavaExecutor with JRE download/management and tar extraction, updates executor factory to route to Java based on parameters/metadata, propagates new options through sources util, re-exports executors in package init, and adds an example script for running Snowflake via Java. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant SourcesUtil as sources.util.get_source
participant ExecUtil as executors.util.get_connector_executor
participant JavaExec as JavaExecutor
participant Docker as DockerExecutor
participant Venv as VenvExecutor
User->>SourcesUtil: get_source(..., use_java, use_java_tar)
SourcesUtil->>ExecUtil: get_connector_executor(..., use_java, use_java_tar)
alt use_java/use_java_tar provided
ExecUtil-->>User: JavaExecutor
else metadata InstallType.JAVA
alt effective_use_java == False
ExecUtil-->>User: DockerExecutor
else effective_use_java == None
alt docker installed
ExecUtil-->>User: DockerExecutor
else
ExecUtil-->>User: JavaExecutor
end
else effective_use_java == True
ExecUtil-->>User: JavaExecutor
end
else non-Java connector
ExecUtil-->>User: Venv/Docker/Declarative
end
sequenceDiagram
participant Caller
participant JavaExec as JavaExecutor
participant Azul as Azul API
participant FS as Filesystem
participant Connector as Java Connector Proc
Caller->>JavaExec: install()
JavaExec->>FS: check JRE/connector dirs
alt JRE missing
JavaExec->>Azul: download JRE (platform/arch/version)
JavaExec->>FS: extract JRE to ~/.airbyte/java/...
end
alt connector tar provided & not extracted
JavaExec->>FS: extract connector tar to ~/.airbyte/connectors/<name>
end
Caller->>JavaExec: execute(args, stdin?)
JavaExec->>Connector: spawn with JAVA_HOME/PATH, flags (--spec/--check/...)
JavaExec-->>Caller: stream stdout lines
alt stdin is message iterator
JavaExec->>Connector: stream messages to stdin
end
Connector-->>JavaExec: exit code
alt non-zero and not terminated
JavaExec-->>Caller: raise AirbyteSubprocessFailedError
else
JavaExec-->>Caller: return
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~40 minutes Possibly related PRs
✨ Finishing Touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (7)
examples/run_source_snowflake_java.py (2)
82-82: Consider extracting the Google Drive file IDAs previously noted, the file ID is hardcoded. For better maintainability, wdyt about moving it to a constant or environment variable?
101-106: Temporary file cleanup could be improvedThe temporary file is created with
delete=Falseand won't be cleaned up if an exception occurs. Even for a demo script, wdyt about adding cleanup or usingdelete=Truesince you're just passing the path?airbyte/_executors/java.py (5)
27-27: Consider using an alias for rich.printShadowing the built-in
rprintinstead?
86-88: Clarify or remove _auto_detect_tar_path methodThis method returns
Noneand its purpose isn't clear. @aaronsteers previously asked about this - could you either expand the docstring to explain its intended use case or remove it if not needed? wdyt?
285-290: Consider additional validation for executable detectionThe current implementation looks for any executable file in bin directories. For added safety, wdyt about validating that found executables are regular files (not symlinks) and are within the expected connector directory bounds?
131-132: Consider using a logging framework instead of print statementsDirect print statements make it harder to control output verbosity. What do you think about using Python's logging module for better control over log levels?
Also applies to: 136-137, 139-140
372-377: Consider making command mapping more flexibleThe hardcoded set of commands that get mapped to
--commandformat might need updates as new commands are added. Would it be worth documenting why only these specific commands need this transformation or making it more configurable? wdyt?
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
airbyte/_executors/__init__.py(1 hunks)airbyte/_executors/java.py(1 hunks)airbyte/_executors/util.py(6 hunks)airbyte/sources/util.py(3 hunks)examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🧬 Code Graph Analysis (4)
examples/run_source_snowflake_java.py (3)
airbyte/sources/util.py (1)
get_source(47-142)airbyte/_executors/java.py (1)
install(302-314)airbyte/sources/base.py (2)
get_available_streams(321-323)select_streams(243-272)
airbyte/_executors/__init__.py (6)
airbyte/_executors/base.py (1)
Executor(159-248)airbyte/_executors/declarative.py (1)
DeclarativeExecutor(39-128)airbyte/_executors/docker.py (1)
DockerExecutor(20-106)airbyte/_executors/java.py (1)
JavaExecutor(44-426)airbyte/_executors/local.py (1)
PathExecutor(14-63)airbyte/_executors/python.py (1)
VenvExecutor(27-342)
airbyte/_executors/java.py (6)
airbyte/_executors/base.py (1)
Executor(159-248)airbyte/_util/telemetry.py (2)
EventState(168-172)log_install_state(298-311)airbyte/version.py (1)
get_version(12-14)airbyte/sources/registry.py (1)
ConnectorMetadata(57-88)airbyte/_message_iterators.py (1)
AirbyteMessageIterator(61-205)airbyte/exceptions.py (2)
AirbyteConnectorInstallationError(344-345)AirbyteSubprocessFailedError(274-277)
airbyte/_executors/util.py (6)
airbyte/_executors/java.py (1)
JavaExecutor(44-426)airbyte/_executors/local.py (1)
PathExecutor(14-63)airbyte/_executors/python.py (1)
VenvExecutor(27-342)airbyte/_util/meta.py (2)
is_docker_installed(168-169)which(154-165)airbyte/_executors/base.py (1)
Executor(159-248)airbyte/sources/registry.py (1)
InstallType(40-46)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Windows)
🔇 Additional comments (5)
airbyte/sources/util.py (1)
62-63: LGTM!The addition of
use_javaanduse_java_tarparameters is well-implemented with clear type hints and comprehensive documentation explaining the fallback behavior and parameter interactions.Also applies to: 116-121, 139-140
airbyte/_executors/__init__.py (1)
4-19: LGTM!The JavaExecutor is properly added to the public API exports following the same pattern as other executors.
airbyte/_executors/util.py (3)
151-151: Print statement changed from stderr to stdoutThe print statement for local executable was changed from stderr to stdout. Was this intentional? It seems unrelated to the Java connector support, wdyt about keeping it consistent with other executors that print to stderr?
175-181: Well-structured Java fallback logic!The fallback logic for Java connectors is clearly documented and implemented with sensible defaults - preferring Docker when available for stability, falling back to Java when Docker isn't available. The effective_use_java calculation properly handles the interaction between use_java and use_java_tar parameters.
Also applies to: 262-276
366-373: LGTM - JavaExecutor instantiationThe JavaExecutor is properly instantiated with all required parameters following the same pattern as other executors.
| with tarfile.open(self.connector_tar_path, "r") as tar: | ||
| tar.extractall(self.connector_dir) |
There was a problem hiding this comment.
Security: Consider validating tar contents before extraction
Using tar.extractall() without validation could be risky if the tar contains absolute paths or path traversal entries (e.g., ../../../etc/passwd). What do you think about adding validation to ensure all paths are relative and within the target directory?
try:
with tarfile.open(self.connector_tar_path, "r") as tar:
+ # Validate tar members before extraction
+ for member in tar.getmembers():
+ if member.name.startswith('/') or '..' in member.name:
+ raise exc.AirbyteConnectorInstallationError(
+ message="Tar file contains unsafe paths",
+ connector_name=self.name,
+ context={"unsafe_path": member.name}
+ )
tar.extractall(self.connector_dir)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| with tarfile.open(self.connector_tar_path, "r") as tar: | |
| tar.extractall(self.connector_dir) | |
| try: | |
| with tarfile.open(self.connector_tar_path, "r") as tar: | |
| - tar.extractall(self.connector_dir) | |
| + # Validate tar members before extraction | |
| + for member in tar.getmembers(): | |
| + if member.name.startswith('/') or '..' in member.name: | |
| + raise exc.AirbyteConnectorInstallationError( | |
| + message="Tar file contains unsafe paths", | |
| + connector_name=self.name, | |
| + context={"unsafe_path": member.name} | |
| + ) | |
| + tar.extractall(self.connector_dir) |
🤖 Prompt for AI Agents
In airbyte/_executors/java.py around lines 260 to 261, the code calls
tar.extractall(self.connector_dir) without validating archive members; modify it
to iterate tar.getmembers(), reject any member with absolute paths (member.name
startswith "/"), reject any member that would escape the target directory when
joined (use os.path.normpath and compare os.path.commonpath of destination and
target), and also handle or skip symlinks appropriately; only extract validated
members (using tar.extract(member, self.connector_dir) or write file content
manually) and raise an error if any unsafe entry is found so extraction cannot
perform path traversal outside self.connector_dir.
…ction - Add use_java and use_java_tar parameters to get_destination() function - Mirror dual-parameter API from get_source() implementation - Add comprehensive parameter documentation for Java execution modes - Update get_connector_executor() call to pass Java parameters - Ensure consistent Java connector support across sources and destinations Requested by: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (5)
airbyte/destinations/util.py (5)
34-36: Signature addition looks good; consider accepting os.PathLike for pathsTo make the API friendlier with libraries returning path-like objects, would you consider accepting os.PathLike[str] in addition to Path/str for use_java and use_java_tar, while keeping current behavior unchanged, wdyt?
Example diff:
@@ -if TYPE_CHECKING: - from pathlib import Path +if TYPE_CHECKING: + from pathlib import Path + from os import PathLike @@ - use_java: Path | str | bool | None = None, - use_java_tar: Path | str | None = None, + use_java: Path | PathLike[str] | str | bool | None = None, + use_java_tar: Path | PathLike[str] | str | None = None,
74-80: Docstring: clarify precedence across executor hintsGiven multiple hints (local_executable, docker_image, use_python, use_java/use_java_tar), shall we add a short “executor selection precedence” note here to reduce ambiguity and align with the new Docker-preferred-then-Java fallback, wdyt? For example:
- local_executable (if set)
- docker_image (if truthy)
- use_python (if truthy or inferred)
- use_java/use_java_tar
- Otherwise: default behavior in get_connector_executor
Also, could we explicitly state here that when use_java_tar is provided, it implies use_java=True unless use_java is explicitly False (to match the underlying implementation)?
44-47: Docstring mentions 'streams' which is not a parameter hereThe Args include a 'streams' description, but get_destination has no such parameter. Shall we remove or relocate this block to avoid confusion, wdyt?
Suggested diff to remove the block:
- streams: list of stream names to select for reading. If set to "*", all streams will be - selected. If not provided, you can set it later via the `select_streams()` or - `select_all_streams()` method.
21-36: Argument creep: consider an ExecutionOptions dataclass to group executor hintsWe’re already suppressing PLR0913, and adding two more knobs increases cognitive load for callers. Would you consider a follow-up refactor to group execution knobs (use_python, local_executable, docker_image, use_host_network, install_if_missing, install_root, use_java, use_java_tar, etc.) into an ExecutionOptions dataclass or similar, keeping this API stable while simplifying signatures, wdyt?
110-119: Prevent unintended JRE downloads in get_noop_destinationSince get_noop_destination is used for perf benchmarking, shall we explicitly disable Java fallback here to avoid any surprise downloads if Docker isn’t available, wdyt?
Example diff:
return get_destination( "destination-dev-null", config={ "test_destination": { "test_destination_type": "SILENT", } }, - docker_image=True, + docker_image=True, + use_java=False, install_if_missing=install_if_missing, )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
airbyte/destinations/util.py(3 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (Fast)
- GitHub Check: Pytest (No Creds)
🔇 Additional comments (1)
airbyte/destinations/util.py (1)
95-97: Executor semantics verified: use_java_tar implies Java, Docker fallback intactI checked the docstring in
airbyte/_executors/util.py(lines 175–181) against the implementation at lines 260–277 and the final JavaExecutor check at lines 366–373. It confirms:
- When
use_java_taris provided anduse_javaisNone,effective_use_javais set toTrue.- If
effective_use_javaisFalse, Docker is chosen; if it’sNone, Docker is used when available (Java otherwise).- Finally,
if use_java or use_java_tarreturns theJavaExecutor.Everything aligns with the intended semantics—no further changes needed, wdyt?
- Replace verbose logging with concise output messages - Consolidate GSM initialization and config retrieval - Streamline connection check and stream discovery flow - Maintain core Java connector functionality with use_java=True - Keep connection check, stream discovery, and 10-record peek demo - Script now meets 50-80 line requirement using GSM pattern Requested by: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (3)
examples/run_source_snowflake_java.py (3)
26-29: Make GCP project configurable via env var for portabilityWould you switch the hard-coded project to an env-provided default so the example runs out-of-the-box in other projects too, wdyt?
Apply this minimal diff:
- secret_mgr = GoogleGSMSecretManager( - project="dataline-integration-testing", - credentials_json=os.environ.get("DEVIN_GCP_SERVICE_ACCOUNT_JSON"), - ) + secret_mgr = GoogleGSMSecretManager( + project=os.environ.get("GCP_PROJECT", "dataline-integration-testing"), + credentials_json=os.environ.get("DEVIN_GCP_SERVICE_ACCOUNT_JSON"), + )
34-36: Expose use_java_tar override via env var (keeps example simple, but more flexible)Since the new API supports use_java_tar, would you allow an env override so users can point at a specific connector tar without changing code, wdyt?
- # Create source with Java execution - source = ab.get_source("source-snowflake", config=config, use_java=True) + # Create source with Java execution (optional tar override via env) + use_java_tar = os.environ.get("SOURCE_SNOWFLAKE_CONNECTOR_TAR") + source = ab.get_source( + "source-snowflake", + config=config, + use_java=True, + use_java_tar=use_java_tar, + )
48-55: Avoid printing full records to reduce potential PII leakage (keep it simple)To keep the example safe, would you print only a summarized view (e.g., keys) or a truncated payload for records, wdyt?
Example (minimal change):
- for record in read_result[selected_stream]: - print(f"Record {records_count + 1}: {record}") + for record in read_result[selected_stream]: + # Print a summarized view to avoid leaking sensitive data + preview = list(record.keys()) if isinstance(record, dict) else str(record)[:200] + print(f"Record {records_count + 1}: {preview}")
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🧬 Code Graph Analysis (1)
examples/run_source_snowflake_java.py (2)
airbyte/secrets/google_gsm.py (1)
fetch_connector_secret(264-298)airbyte/sources/base.py (2)
get_available_streams(321-323)select_streams(243-272)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (No Creds)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (Fast)
🔇 Additional comments (2)
examples/run_source_snowflake_java.py (2)
1-67: LGTM overall — clear, concise example highlighting Java pathThe flow (fetch config → construct source with use_java=True → check → enumerate → read) is easy to follow and aligns with the new Java executor UX. Nice work!
38-41: Explicit discovery already handled by get_available_streams()The call to
source.get_available_streams()accesses thediscovered_catalogproperty, which lazily invokes_discover()on first access. That means discovery is guaranteed before listing streams, so an explicitsource.discover()call isn’t needed—wdyt?
- Fix missing return statement in get_connector_executor Java block - Ensures JavaExecutor is returned instead of falling through to VenvExecutor - Fixes use_java_tar parameter functionality in example script - Script now properly downloads tar and uses Java connector execution Fixes issue reported by: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
- Add tar download functionality back to source-snowflake example - Replace use_java_tar='TODO' with actual downloaded tar file path - Update get_records call to use select_streams and read methods - Script now properly downloads tar from Google Drive and uses JavaExecutor Addresses feedback from: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (4)
examples/run_source_snowflake_java.py (1)
26-26: Avoid hardcoded Google Drive file IDCan we source the Google Drive file ID from an environment variable with a sane default to avoid the magic string and ease future rotations, wdyt?
- file_id = "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR" + file_id = os.getenv("SOURCE_SNOWFLAKE_CONNECTOR_FILE_ID", "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR")airbyte/_executors/java.py (3)
28-28: Avoid shadowing built-in printMinor: importing rich.print as print can be confusing across the module. Would aliasing it (e.g., rprint) reduce ambiguity and ease future refactors, wdyt?
-from rich import print # noqa: A004 # Allow shadowing the built-in +from rich import print as rprint # Avoid shadowing the built-inIf you like this, I can follow through by replacing the occurrences accordingly.
121-143: Optional: centralize user-visible output via loggingWould you consider routing these user-facing prints through logging (or a centralized output helper) for better control over verbosity in library contexts, wdyt?
259-264: Tar extraction without validation (path traversal / symlink risks)Extracting arbitrary tar contents with extractall is unsafe. Could we validate members to prevent path traversal and deny symlinks/hardlinks before extracting, wdyt?
- try: - with tarfile.open(self.connector_tar_path, "r") as tar: - tar.extractall(self.connector_dir) + try: + with tarfile.open(self.connector_tar_path, "r") as tar: + target_root = self.connector_dir.resolve() + safe_members: list[tarfile.TarInfo] = [] + for m in tar.getmembers(): + # Disallow links outright for safety + if m.islnk() or m.issym(): + raise exc.AirbyteConnectorInstallationError( + message="Tar contains symlinks or hardlinks, which are not allowed.", + connector_name=self.name, + context={"member": m.name}, + ) + # Normalize the destination and ensure it's under target_root + dest_path = (target_root / m.name).resolve() + if not str(dest_path).startswith(str(target_root)): + raise exc.AirbyteConnectorInstallationError( + message="Tar contains an unsafe path (path traversal detected).", + connector_name=self.name, + context={"member": m.name}, + ) + safe_members.append(m) + tar.extractall(self.connector_dir, members=safe_members)Would you like me to also add a unit test fixture to assert that “../evil” entries are rejected?
🧹 Nitpick comments (3)
examples/run_source_snowflake_java.py (2)
14-16: Ensure temp tar file is cleaned up on exitSince we’re using delete=False, can we register a cleanup to avoid accumulating temp files if the script errors, wdyt?
@@ -import os +import os +import atexit @@ - tar_path = download_snowflake_tar() + tar_path = download_snowflake_tar() + atexit.register(lambda p=tar_path: p.unlink(missing_ok=True)) @@ except Exception as e: print(f"❌ Error: {e}") raiseAlso applies to: 52-56, 88-91
79-84: Avoid materializing all records when samplingTiny nit: list(...) will materialize the whole stream before slicing. For demos, a bounded loop avoids unnecessary memory. Shall we iterate and break after 10, wdyt?
- read_result = source.read() - records = list(read_result[selected_stream])[:10] - print(f"✅ Read {len(records)} records using Java connector!") + read_result = source.read() + count = 0 + for _ in read_result[selected_stream]: + count += 1 + if count >= 10: + break + print(f"✅ Read {count} records using Java connector!")airbyte/_executors/java.py (1)
8-13: Docstring nit: parameter mismatch in fallback descriptionThe bullet “When use_java_tar=False” doesn’t align with the function signature (use_java_tar is Path | str | None). Should this be “When use_java is False: …” to match the actual switch, wdyt?
- - When use_java_tar=False: Java execution is explicitly disabled, fallback to Docker + - When use_java is False: Java execution is explicitly disabled, fallback to Docker
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
airbyte/_executors/java.py(1 hunks)airbyte/_executors/util.py(6 hunks)examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🧬 Code Graph Analysis (3)
examples/run_source_snowflake_java.py (4)
airbyte/secrets/google_gsm.py (1)
fetch_connector_secret(264-298)airbyte/cloud/connections.py (1)
source(81-90)airbyte/sources/util.py (1)
get_source(47-142)airbyte/sources/base.py (2)
get_available_streams(321-323)select_streams(243-272)
airbyte/_executors/util.py (7)
airbyte/_executors/java.py (2)
JavaExecutor(45-426)ensure_installation(324-346)airbyte/_executors/python.py (2)
VenvExecutor(27-342)ensure_installation(246-337)airbyte/_util/meta.py (2)
is_docker_installed(168-169)which(154-165)airbyte/_executors/base.py (2)
Executor(159-248)ensure_installation(226-228)airbyte/sources/registry.py (1)
InstallType(40-46)airbyte/_executors/docker.py (1)
ensure_installation(34-52)airbyte/_util/telemetry.py (2)
log_install_state(298-311)EventState(168-172)
airbyte/_executors/java.py (6)
airbyte/_executors/base.py (1)
Executor(159-248)airbyte/_util/telemetry.py (2)
EventState(168-172)log_install_state(298-311)airbyte/version.py (1)
get_version(12-14)airbyte/sources/registry.py (1)
ConnectorMetadata(57-88)airbyte/_message_iterators.py (1)
AirbyteMessageIterator(61-205)airbyte/exceptions.py (2)
AirbyteConnectorInstallationError(344-345)AirbyteSubprocessFailedError(274-277)
🪛 ast-grep (0.38.6)
examples/run_source_snowflake_java.py
[warning] 41-41: The function mktemp is deprecated. When using this function, it is possible for an attacker to modify the created file before the filename is returned. Use NamedTemporaryFile() instead and pass it the delete=False parameter.
Context: tempfile.mktemp(suffix=".tar")
Note: [CWE-377]: Insecure Temporary File [OWASP A01:2021]: Broken Access Control [REFERENCES]
https://docs.python.org/3/library/tempfile.html#tempfile.mktemp
https://owasp.org/Top10/A01_2021-Broken_Access_Control
(avoid-mktemp-python)
🪛 GitHub Check: CodeQL
examples/run_source_snowflake_java.py
[failure] 42-42: Insecure temporary file
Call to deprecated function tempfile.mktemp may be insecure.
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (Fast)
- GitHub Check: Pytest (No Creds)
🔇 Additional comments (7)
airbyte/_executors/util.py (7)
151-151: Local executor log routing LGTMSwitching this status message to stdout looks good and aligns with the rest of the factory output.
169-172: API surface extension for Java params looks consistentAdding use_java and use_java_tar here matches the PR’s stated dual-parameter API and keeps their types flexible. Nice.
175-181: Java fallback logic reads clearly and matches intentThe documented precedence is clear and matches the implementation below. Thanks for spelling this out inline.
206-221: Conflict validation message includes Java flagsGood catch including 'use_java' and 'use_java_tar' in the mutual-exclusion error. This will save users time.
263-276: Potential dead path when Docker is absent and no TAR is providedOne edge case: when Docker is not installed and effective_use_java is None, the code sets use_java=True (Lines 269-274). With the current JavaExecutor, _auto_detect_tar_path() returns None and install() will fail when extracting the tar. Until auto-detection is implemented, should we either (a) avoid auto-enabling Java without a tar, or (b) ensure JavaExecutor can reliably auto-detect a tar or an already-extracted connector, wdyt?
I can help implement the auto-detect in JavaExecutor so this path is robust. Would you like me to draft it to look for env overrides and common patterns under ~/.airbyte or Downloads?
366-377: Short-circuit to JavaExecutor is clean and respects install_if_missingThe creation flow and ensure_installation call look correct and consistent with Python/Docker paths.
381-397: Venv path: ensure_installation on success and explicit returnThis aligns with the other executors and gives consistent behavior for auto-fix. LGTM.
| if use_java_tar is not None and use_java is None: | ||
| use_java = True | ||
|
|
||
| self.use_java = use_java | ||
| self.connector_tar_path = Path(use_java_tar) if use_java_tar else None | ||
| self.java_version = "21" | ||
| self.airbyte_home = Path.home() / ".airbyte" | ||
| self.java_cache_dir = self.airbyte_home / "java" | ||
|
|
||
| self.os_name, self.arch = self._detect_platform() | ||
| self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}" | ||
|
|
||
| self.connector_dir = self.airbyte_home / "connectors" / self.name | ||
|
|
||
| if self.connector_tar_path is None and self.use_java: | ||
| self.connector_tar_path = self._auto_detect_tar_path() | ||
|
|
There was a problem hiding this comment.
Auto-detect TAR path not implemented – breaks fallback when Docker is absent
_auto_detect_tar_path returns None, which will cause install() to fail when no explicit use_java_tar is provided. Could we implement a minimal auto-detection pass (env overrides + common locations) to make the Docker->Java fallback robust, wdyt?
def _auto_detect_tar_path(self) -> Path | None:
- """Auto-detect the connector tar file path."""
- return None
+ """Auto-detect the connector tar file path.
+
+ Strategy:
+ 1) Env overrides:
+ - AIRBYTE_{NAME}_TAR (dashes -> underscores, uppercased)
+ - AIRBYTE_JAVA_CONNECTOR_TAR
+ 2) Common local locations:
+ - Current working directory
+ - ~/Downloads
+ We look for patterns like '{name}*.tar', '{name}*.tar.gz', '{name}*.tgz'.
+ """
+ # 1) Env overrides
+ env_vars = [
+ f"AIRBYTE_{self.name.upper().replace('-', '_')}_TAR",
+ "AIRBYTE_JAVA_CONNECTOR_TAR",
+ ]
+ for var in env_vars:
+ val = os.environ.get(var)
+ if val:
+ path = Path(val).expanduser().resolve()
+ if path.exists():
+ print(f"🔎 Using TAR from env {var}: {path}")
+ return path
+
+ # 2) Common locations/patterns
+ candidates: list[Path] = []
+ search_roots = [Path.cwd(), Path.home() / "Downloads"]
+ patterns = (f"{self.name}*.tar", f"{self.name}*.tar.gz", f"{self.name}*.tgz")
+ for root in search_roots:
+ for pattern in patterns:
+ candidates.extend(root.glob(pattern))
+
+ if candidates:
+ chosen = sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[0]
+ print(f"🔎 Auto-detected TAR: {chosen}")
+ return chosen
+ return NoneWould you also like me to add a helpful error message in install() when self.connector_tar_path is still None to guide users toward setting use_java_tar?
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if use_java_tar is not None and use_java is None: | |
| use_java = True | |
| self.use_java = use_java | |
| self.connector_tar_path = Path(use_java_tar) if use_java_tar else None | |
| self.java_version = "21" | |
| self.airbyte_home = Path.home() / ".airbyte" | |
| self.java_cache_dir = self.airbyte_home / "java" | |
| self.os_name, self.arch = self._detect_platform() | |
| self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}" | |
| self.connector_dir = self.airbyte_home / "connectors" / self.name | |
| if self.connector_tar_path is None and self.use_java: | |
| self.connector_tar_path = self._auto_detect_tar_path() | |
| def _auto_detect_tar_path(self) -> Path | None: | |
| """Auto-detect the connector tar file path. | |
| Strategy: | |
| 1) Env overrides: | |
| - AIRBYTE_{NAME}_TAR (dashes -> underscores, uppercased) | |
| - AIRBYTE_JAVA_CONNECTOR_TAR | |
| 2) Common local locations: | |
| - Current working directory | |
| - ~/Downloads | |
| We look for patterns like '{name}*.tar', '{name}*.tar.gz', '{name}*.tgz'. | |
| """ | |
| # 1) Env overrides | |
| env_vars = [ | |
| f"AIRBYTE_{self.name.upper().replace('-', '_')}_TAR", | |
| "AIRBYTE_JAVA_CONNECTOR_TAR", | |
| ] | |
| for var in env_vars: | |
| val = os.environ.get(var) | |
| if val: | |
| path = Path(val).expanduser().resolve() | |
| if path.exists(): | |
| print(f"🔎 Using TAR from env {var}: {path}") | |
| return path | |
| # 2) Common locations/patterns | |
| candidates: list[Path] = [] | |
| search_roots = [Path.cwd(), Path.home() / "Downloads"] | |
| patterns = ( | |
| f"{self.name}*.tar", | |
| f"{self.name}*.tar.gz", | |
| f"{self.name}*.tgz", | |
| ) | |
| for root in search_roots: | |
| for pattern in patterns: | |
| candidates.extend(root.glob(pattern)) | |
| if candidates: | |
| chosen = sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[0] | |
| print(f"🔎 Auto-detected TAR: {chosen}") | |
| return chosen | |
| return None |
| self.java_version = "21" | ||
| self.airbyte_home = Path.home() / ".airbyte" | ||
| self.java_cache_dir = self.airbyte_home / "java" | ||
|
|
||
| self.os_name, self.arch = self._detect_platform() | ||
| self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}" | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Cache directory should include Java version
Storing the JRE cache under only os/arch risks stale JRE reuse across version bumps. Shall we include the Java major version in the cache key, wdyt?
- self.os_name, self.arch = self._detect_platform()
- self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}"
+ self.os_name, self.arch = self._detect_platform()
+ self.jre_dir = self.java_cache_dir / f"jre-{self.java_version}-{self.os_name}-{self.arch}"📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| self.java_version = "21" | |
| self.airbyte_home = Path.home() / ".airbyte" | |
| self.java_cache_dir = self.airbyte_home / "java" | |
| self.os_name, self.arch = self._detect_platform() | |
| self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}" | |
| self.java_version = "21" | |
| self.airbyte_home = Path.home() / ".airbyte" | |
| self.java_cache_dir = self.airbyte_home / "java" | |
| self.os_name, self.arch = self._detect_platform() | |
| self.jre_dir = self.java_cache_dir / f"jre-{self.java_version}-{self.os_name}-{self.arch}" |
🤖 Prompt for AI Agents
In airbyte/_executors/java.py around lines 75 to 81, the JRE cache path only
uses os and arch which can cause reuse of stale JREs when the Java major version
changes; change the jre_dir construction to include the java_version (e.g.
append or incorporate the major version into the directory name) so the cache
key becomes os-arch-java{version} (or similar), and ensure any downstream
references to jre_dir continue to work with the new path.
| for bin_dir in self.connector_dir.rglob("bin"): | ||
| for executable in bin_dir.iterdir(): | ||
| if ( | ||
| executable.is_file() and executable.stat().st_mode & 0o111 | ||
| ): # Check if executable | ||
| return executable | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Executable discovery: tighten selection and avoid symlinks
Returning the first executable found under any bin/ dir may pick the wrong binary. Shall we prefer a name match, avoid symlinks, and ensure paths remain within connector_dir, wdyt?
- for bin_dir in self.connector_dir.rglob("bin"):
- for executable in bin_dir.iterdir():
- if (
- executable.is_file() and executable.stat().st_mode & 0o111
- ): # Check if executable
- return executable
+ for bin_dir in self.connector_dir.rglob("bin"):
+ # Prefer a direct name match first
+ candidate = bin_dir / self.name
+ if candidate.exists() and candidate.is_file() and (candidate.stat().st_mode & 0o111):
+ return candidate.resolve()
+ # Fallback: any executable that is a regular file, not a symlink, and within connector_dir
+ for executable in bin_dir.iterdir():
+ try:
+ resolved = executable.resolve()
+ except OSError:
+ continue
+ if (
+ executable.is_file()
+ and not executable.is_symlink()
+ and (executable.stat().st_mode & 0o111)
+ and str(resolved).startswith(str(self.connector_dir.resolve()))
+ ):
+ return resolved📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for bin_dir in self.connector_dir.rglob("bin"): | |
| for executable in bin_dir.iterdir(): | |
| if ( | |
| executable.is_file() and executable.stat().st_mode & 0o111 | |
| ): # Check if executable | |
| return executable | |
| for bin_dir in self.connector_dir.rglob("bin"): | |
| # Prefer a direct name match first | |
| candidate = bin_dir / self.name | |
| if candidate.exists() and candidate.is_file() and (candidate.stat().st_mode & 0o111): | |
| return candidate.resolve() | |
| # Fallback: any executable that is a regular file, not a symlink, and within connector_dir | |
| for executable in bin_dir.iterdir(): | |
| try: | |
| resolved = executable.resolve() | |
| except OSError: | |
| continue | |
| if ( | |
| executable.is_file() | |
| and not executable.is_symlink() | |
| and (executable.stat().st_mode & 0o111) | |
| and str(resolved).startswith(str(self.connector_dir.resolve())) | |
| ): | |
| return resolved |
🤖 Prompt for AI Agents
In airbyte/_executors/java.py around lines 285 to 291, the current loop returns
the first file with an executable bit under any bin/ directory which can pick
the wrong binary and follow symlinks outside the connector; change the selection
to (1) prefer a file whose name exactly matches the expected executable name
(fall back to other executables only if no exact match), (2) skip symlinks
(ensure Path.is_symlink() is False and use Path.resolve() for real paths), (3)
verify the resolved path is inside self.connector_dir (e.g., compare parents or
use commonpath), and (4) ensure it is a regular file with the executable bit
before returning; iterate deterministically (sorted) so behavior is stable.
…ning - Extract UUID from virus scan warning HTML form - Use correct download URL with confirm=t parameter - Successfully downloads 137MB valid tar file instead of HTML - Java connector execution now works correctly - Connection check fails due to OAuth vs username/password credential mismatch Fixes Google Drive download issue reported by: @aaronsteers Co-Authored-By: AJ Steers <[email protected]>
- Remove debug comments and improve code formatting - Ensure proper UUID extraction from virus scan warning - Maintain 67-line script length requirement - All linting and type checking passes Final fix for Google Drive tar download functionality. Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (2)
examples/run_source_snowflake_java.py (2)
25-25: Avoid a magic string for the Google Drive file IDCould we make the Drive file ID configurable via an env var with a sensible default, to ease updates without editing code, wdyt?
- file_id = "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR" + file_id = os.getenv("SOURCE_SNOWFLAKE_CONNECTOR_FILE_ID", "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR")
41-43: Fix insecure temporary file creation (mktemp is deprecated and unsafe)This is flagged by CodeQL and ast-grep. Let's switch to NamedTemporaryFile(delete=False) to avoid TOCTOU risks, wdyt?
- temp_file = Path(tempfile.mktemp(suffix=".tar")) - temp_file.write_bytes(response.content) - return temp_file + with tempfile.NamedTemporaryFile(delete=False, suffix=".tar") as tf: + tf.write(response.content) + return Path(tf.name)
🧹 Nitpick comments (4)
examples/run_source_snowflake_java.py (4)
27-39: Add timeouts and raise_for_status() for HTTP callsWould you add explicit timeouts and raise_for_status() to fail fast on network errors and surface HTTP failures more clearly, wdyt?
- response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}") + response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}", timeout=30) + response.raise_for_status() @@ - response = session.get( - f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}" - ) + response = session.get( + f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}", + timeout=30, + ) + response.raise_for_status()
31-39: Drive “virus scan” handling is brittle across locales/layoutsThe string check for "virus scan warning" and parsing a specific hidden input may break if Google changes markup or language. Would you consider a more resilient approach (e.g., look for a “confirm” token in cookies or any hidden input named “confirm”, with a generic fallback), wdyt?
56-59: Parameterize project and validate credentials env varTo make the demo more portable and fail fast with a clear message, could we source the project from an env var (with default) and assert the creds var is set, wdyt?
- secret_mgr = GoogleGSMSecretManager( - project="dataline-integration-testing", - credentials_json=os.environ.get("DEVIN_GCP_SERVICE_ACCOUNT_JSON"), - ) + project_id = os.getenv("GCP_PROJECT_ID", "dataline-integration-testing") + creds_json = os.environ.get("DEVIN_GCP_SERVICE_ACCOUNT_JSON") + if not creds_json: + raise RuntimeError("DEVIN_GCP_SERVICE_ACCOUNT_JSON is required to run this example.") + secret_mgr = GoogleGSMSecretManager( + project=project_id, + credentials_json=creds_json, + )
62-63: Confirm printing of potentially sensitive config fieldsAre we okay printing the Snowflake account identifier in logs for this example? If not, we could redact or log a truncated form, wdyt?
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🪛 ast-grep (0.38.6)
examples/run_source_snowflake_java.py
[warning] 40-40: The function mktemp is deprecated. When using this function, it is possible for an attacker to modify the created file before the filename is returned. Use NamedTemporaryFile() instead and pass it the delete=False parameter.
Context: tempfile.mktemp(suffix=".tar")
Note: [CWE-377]: Insecure Temporary File [OWASP A01:2021]: Broken Access Control [REFERENCES]
https://docs.python.org/3/library/tempfile.html#tempfile.mktemp
https://owasp.org/Top10/A01_2021-Broken_Access_Control
(avoid-mktemp-python)
🪛 GitHub Check: CodeQL
examples/run_source_snowflake_java.py
[failure] 41-41: Insecure temporary file
Call to deprecated function tempfile.mktemp may be insecure.
🪛 GitHub Actions: Run Linters
examples/run_source_snowflake_java.py
[error] 79-79: mypy: Statement is unreachable [unreachable]. (Command: 'poetry run mypy .')
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (Fast)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (No Creds)
🔇 Additional comments (1)
examples/run_source_snowflake_java.py (1)
1-10: Concise, readable demo scaffold — LGTMThe header, usage notes, and docstring set clear expectations for a demo script and align with our guidance for examples. Nice.
- Fix CodeQL security alert by using tempfile.mkstemp() instead of mktemp() - Prevents race condition vulnerabilities in temporary file creation - Maintain same functionality for Google Drive tar download - Add proper file descriptor cleanup Fixes CodeQL high severity security alert in PR #746 Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (3)
examples/run_source_snowflake_java.py (3)
26-26: Avoid hardcoding the Google Drive file ID (use an env override).Would you read the ID from an environment variable with a sensible default to make updates easier, wdyt?
Apply this diff:
- file_id = "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR" + file_id = os.getenv("SOURCE_SNOWFLAKE_CONNECTOR_FILE_ID", "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR")
40-42: Fix insecure temporary file creation (mktemp is deprecated and unsafe).tempfile.mktemp() is insecure and flagged by CodeQL/OWASP. Could we switch to NamedTemporaryFile(delete=False) to safely create the tar file, wdyt?
Apply this diff:
- temp_file = Path(tempfile.mktemp(suffix=".tar")) - temp_file.write_bytes(response.content) - return temp_file + with tempfile.NamedTemporaryFile(delete=False, suffix=".tar") as tf: + tf.write(response.content) + return Path(tf.name)
76-91: Remove the early return; gate the full demo behind an env flag to fix mypy “unreachable” error.The unconditional return makes following code unreachable and trips mypy. Shall we make the “full run” optional via an env flag and drop the early return, wdyt?
Apply this diff:
- _ = source.config_spec - print("✅ Config spec retrieved successfully!") - - return # This is as far as we can go for now. - - # TODO: Fix this part. Connector doesn't seem to get the config properly. - source.check() - print("✅ Connection check passed") - stream_names = source.get_available_streams() - print(f"📊 Found {len(stream_names)} streams") - - selected_stream = stream_names[0] - source.select_streams([selected_stream]) - read_result = source.read() - records = list(read_result[selected_stream])[:10] - print(f"✅ Read {len(records)} records using Java connector!") + _ = source.config_spec + print("✅ Config spec retrieved successfully!") + + # Optional: run full demo (connection check and a short read) if explicitly enabled. + if os.getenv("RUN_SNOWFLAKE_JAVA_DEMO_FULL") == "1": + source.check() + print("✅ Connection check passed") + stream_names = source.get_available_streams() + print(f"📊 Found {len(stream_names)} streams") + + if stream_names: + selected_stream = stream_names[0] + source.select_streams([selected_stream]) + read_result = source.read() + records = list(read_result[selected_stream])[:10] + print(f"✅ Read {len(records)} records using Java connector!") + else: + print("❌ No streams found") + else: + print("ℹ️ Skipping full run; set RUN_SNOWFLAKE_JAVA_DEMO_FULL=1 to run connection check and read.")
🧹 Nitpick comments (1)
examples/run_source_snowflake_java.py (1)
29-38: Use a context manager for the session and add a timeout to HTTP calls.This avoids leaking the session and prevents hanging on slow responses. Keeping it simple for the example, could we add a short timeout and use a with-block, wdyt?
Apply this diff:
- # Create session and get initial response - session = requests.Session() - response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}") - - if "virus scan warning" in response.text.lower(): - uuid_match = re.search(r'name="uuid" value="([^"]+)"', response.text) - uuid_value = uuid_match.group(1) if uuid_match else "" - - response = session.get( - f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}" - ) + # Create session and get initial response + with requests.Session() as session: + response = session.get( + f"https://drive.google.com/uc?export=download&id={file_id}", + timeout=60, + ) + + if "virus scan warning" in response.text.lower(): + uuid_match = re.search(r'name="uuid" value="([^"]+)"', response.text) + uuid_value = uuid_match.group(1) if uuid_match else "" + + response = session.get( + f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}", + timeout=60, + )
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🧬 Code Graph Analysis (1)
examples/run_source_snowflake_java.py (4)
airbyte/secrets/google_gsm.py (3)
GoogleGSMSecretManager(75-298)GSMSecretHandle(54-72)fetch_connector_secret(264-298)airbyte/cloud/connections.py (1)
source(81-90)airbyte/sources/util.py (1)
get_source(47-142)airbyte/sources/base.py (2)
get_available_streams(321-323)select_streams(243-272)
🪛 ast-grep (0.38.6)
examples/run_source_snowflake_java.py
[warning] 39-39: The function mktemp is deprecated. When using this function, it is possible for an attacker to modify the created file before the filename is returned. Use NamedTemporaryFile() instead and pass it the delete=False parameter.
Context: tempfile.mktemp(suffix=".tar")
Note: [CWE-377]: Insecure Temporary File [OWASP A01:2021]: Broken Access Control [REFERENCES]
https://docs.python.org/3/library/tempfile.html#tempfile.mktemp
https://owasp.org/Top10/A01_2021-Broken_Access_Control
(avoid-mktemp-python)
🪛 GitHub Check: CodeQL
examples/run_source_snowflake_java.py
[failure] 40-40: Insecure temporary file
Call to deprecated function tempfile.mktemp may be insecure.
🪛 GitHub Actions: Run Linters
examples/run_source_snowflake_java.py
[error] 1-1: Command 'poetry run ruff format --check .' failed: Ruff format check would reformat 1 file: examples/run_source_snowflake_java.py.
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (No Creds)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (Fast)
🔇 Additional comments (1)
examples/run_source_snowflake_java.py (1)
1-97: Formatting applied successfullyThe
examples/run_source_snowflake_java.pyfile has been reformatted and now passesruff format --check. CI should be green.Would you also update your
.ruff.tomlto move the top-levelselectoption intolint.selectto silence the deprecation warning? wdyt?
…hub.com/airbytehq/PyAirbyte into devin/1754946348-java-connector-support
- Fix formatting issues introduced by security fix - Maintain tempfile.mkstemp() security improvement - Resolve Ruff Format Check CI failure Addresses formatting CI failure in PR #746 Co-Authored-By: AJ Steers <[email protected]>
- Replace tar.getmembers() + tar.extract() with streaming iteration - Use tar.extractfile() and manual file writing for streaming compatibility - Maintain strip-components behavior for JRE directory structure - Preserve file permissions during extraction - Fix 'seeking backwards is not allowed' StreamError Resolves JRE extraction issue in Java connector support Co-Authored-By: AJ Steers <[email protected]>
Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (9)
examples/run_source_snowflake_java.py (3)
35-35: Externalize the Google Drive file ID via env var to avoid magic stringWould you switch to an env var for easier rotation and local overrides, wdyt?
- file_id = "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR" + file_id = os.getenv("SOURCE_SNOWFLAKE_CONNECTOR_FILE_ID", "1S0yMrdhs2TLu5u1yvj-52kaeagRAG9ZR")
49-53: Prefer NamedTemporaryFile to simplify and ensure safe temp handlingmkstemp works, but NamedTemporaryFile(delete=False) is simpler and avoids manual fd management. Want to switch, wdyt?
- fd, temp_path = tempfile.mkstemp(suffix=".tar") - temp_file = Path(temp_path) - os.close(fd) - temp_file.write_bytes(response.content) - return temp_file + with tempfile.NamedTemporaryFile(delete=False, suffix=".tar") as tf: + tf.write(response.content) + return Path(tf.name)
93-106: Fix mypy “unreachable” by removing early return and gating full demo behind an env flagThe unconditional return makes the rest unreachable and fails mypy. Would you gate the “full run” behind an env flag instead, wdyt?
- return # This is as far as we can go for now. - - # TODO: Fix this part. Connector doesn't seem to get the config properly. - source.check() - print("✅ Connection check passed") - stream_names = source.get_available_streams() - print(f"📊 Found {len(stream_names)} streams") - - selected_stream = stream_names[0] - source.select_streams([selected_stream]) - read_result = source.read() - records = list(read_result[selected_stream])[:10] - print(f"✅ Read {len(records)} records using Java connector!") + # Optional: run full demo (connection check and a short read) if explicitly enabled. + if os.getenv("RUN_SNOWFLAKE_JAVA_DEMO_FULL") == "1": + # TODO: Fix this part. Connector doesn't seem to get the config properly. + source.check() + print("✅ Connection check passed") + stream_names = source.get_available_streams() + print(f"📊 Found {len(stream_names)} streams") + + if stream_names: + selected_stream = stream_names[0] + source.select_streams([selected_stream]) + read_result = source.read() + records = list(read_result[selected_stream])[:10] + print(f"✅ Read {len(records)} records using Java connector!") + else: + print("❌ No streams found") + else: + print("ℹ️ Skipping full run; set RUN_SNOWFLAKE_JAVA_DEMO_FULL=1 to run connection check and read.")airbyte/_executors/java.py (6)
79-81: Include Java version in JRE cache dir to avoid stale reuse across upgradesStashing by only os/arch risks mixing versions. Would you include java_version in the cache key, wdyt?
- self.jre_dir = self.java_cache_dir / f"{self.os_name}-{self.arch}" + self.jre_dir = self.java_cache_dir / f"jre-{self.java_version}-{self.os_name}-{self.arch}"
384-390: Centralize command mapping set for clarityWould you extract the mapped command set to a constant to ease future additions, wdyt?
- mapped_args = [] - for arg in args: - if arg in {"spec", "check", "discover", "read"}: + COMMANDS_TO_MAP = {"spec", "check", "discover", "read"} + mapped_args = [] + for arg in args: + if arg in COMMANDS_TO_MAP: mapped_args.append(f"--{arg}") else: mapped_args.append(arg)
28-29: Avoid shadowing built-in print with rich.printTo reduce surprises for contributors, would you alias it (e.g., rprint) and follow-up to update usages, wdyt?
-from rich import print # noqa: A004 # Allow shadowing the built-in +from rich import print as rprint # Avoid shadowing the built-inNote: This would require replacing print(...) calls with rprint(...). I can follow up with a patch if you’d like.
271-274: Harden tar extraction against path traversal and unsafe entriesextractall without validation is risky. Would you validate members to ensure no absolute paths, “..” segments, or unsafe symlinks before extraction, wdyt?
- with tarfile.open(self.connector_tar_path, "r") as tar: - tar.extractall(self.connector_dir) + with tarfile.open(self.connector_tar_path, "r") as tar: + safe_members = [] + root = self.connector_dir.resolve() + for member in tar.getmembers(): + name = member.name + # Disallow absolute paths and parent traversal + if name.startswith("/") or ".." in Path(name).parts: + raise exc.AirbyteConnectorInstallationError( + message="Connector tar contains unsafe path.", + connector_name=self.name, + context={"unsafe_path": name}, + ) + # Optionally skip symlinks if not desired: + # if member.issym() or member.islnk(): continue + safe_members.append(member) + tar.extractall(self.connector_dir, members=safe_members)
297-303: Tighten executable discovery to avoid wrong picks and symlinksReturning the first executable under any bin/ is brittle. Shall we prefer an exact name match, avoid symlinks, and ensure the resolved path stays within connector_dir, wdyt?
- for bin_dir in self.connector_dir.rglob("bin"): - for executable in bin_dir.iterdir(): - if ( - executable.is_file() and executable.stat().st_mode & 0o111 - ): # Check if executable - return executable + for bin_dir in self.connector_dir.rglob("bin"): + # Prefer exact name match first + preferred = bin_dir / self.name + if preferred.exists() and preferred.is_file() and (preferred.stat().st_mode & 0o111): + return preferred.resolve() + # Fallback: any safe executable + for executable in sorted(bin_dir.iterdir()): + try: + resolved = executable.resolve() + except OSError: + continue + if ( + executable.is_file() + and not executable.is_symlink() + and (executable.stat().st_mode & 0o111) + and str(resolved).startswith(str(self.connector_dir.resolve())) + ): + return resolved
87-90: Implement auto-detect tar path for robust Docker→Java fallback_auto_detect_tar_path currently returns None. Shall we implement env and common-location detection so Java fallback works out-of-the-box, wdyt?
def _auto_detect_tar_path(self) -> Path | None: - """Auto-detect the connector tar file path.""" - return None + """Auto-detect the connector tar file path. + + Strategy: + 1) Env overrides: + - AIRBYTE_{NAME}_TAR (dashes -> underscores, uppercased) + - AIRBYTE_JAVA_CONNECTOR_TAR + 2) Common local locations: + - Current working directory + - ~/Downloads + """ + # 1) Env overrides + env_vars = [ + f"AIRBYTE_{self.name.upper().replace('-', '_')}_TAR", + "AIRBYTE_JAVA_CONNECTOR_TAR", + ] + for var in env_vars: + val = os.environ.get(var) + if val: + path = Path(val).expanduser().resolve() + if path.exists(): + print(f"🔎 Using TAR from env {var}: {path}") + return path + + # 2) Common locations/patterns + candidates: list[Path] = [] + search_roots = [Path.cwd(), Path.home() / "Downloads"] + patterns = (f"{self.name}*.tar", f"{self.name}*.tar.gz", f"{self.name}*.tgz") + for root in search_roots: + for pattern in patterns: + candidates.extend(root.glob(pattern)) + if candidates: + chosen = sorted(candidates, key=lambda p: p.stat().st_mtime, reverse=True)[0] + print(f"🔎 Auto-detected TAR: {chosen}") + return chosen + return None
🧹 Nitpick comments (4)
examples/run_source_snowflake_java.py (2)
39-47: Add request timeouts to make the demo resilient to network hangsQuick nit: adding a timeout keeps the demo snappy and avoids indefinite hangs if Google Drive stalls, wdyt?
- response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}") + response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}", timeout=60) @@ - response = session.get( - f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}" - ) + response = session.get( + f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}", + timeout=60, + )
56-64: Broaden return type of get_connector_config to dict[str, Any]parse_json likely returns heterogeneous values, so dict[str, str] is too narrow and may trip mypy later. Shall we widen it, wdyt?
+from typing import Any @@ -def get_connector_config(connector_name: str) -> dict[str, str]: +def get_connector_config(connector_name: str) -> dict[str, Any]:Also applies to: 12-18
airbyte/_executors/java.py (2)
380-381: Use os.pathsep when prepending to PATHMinor portability nit: os.pathsep is safer than a hardcoded colon. Even if we currently target POSIX, this is trivial, wdyt?
- env["PATH"] = f"{java_home / 'bin'}:{env.get('PATH', '')}" + env["PATH"] = f"{java_home / 'bin'}{os.pathsep}{env.get('PATH', '')}"
197-205: Ensure streamed response decompresses correctly during JRE extractionTo avoid issues with compressed streams, would you set decode_content before handing raw to tarfile, wdyt?
- response = requests.get(jre_url, stream=True, timeout=300) + response = requests.get(jre_url, stream=True, timeout=300) response.raise_for_status() - with tarfile.open(fileobj=response.raw, mode="r|gz") as tar: + # Ensure urllib3 handles gzip decompression + response.raw.decode_content = True + with tarfile.open(fileobj=response.raw, mode="r|gz") as tar: self._extract_jre_with_strip_components(tar)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
airbyte/_executors/java.py(1 hunks)examples/run_source_snowflake_java.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_source_snowflake_java.py
🧬 Code Graph Analysis (2)
examples/run_source_snowflake_java.py (2)
airbyte/secrets/google_gsm.py (3)
GoogleGSMSecretManager(75-298)GSMSecretHandle(54-72)fetch_connector_secret(264-298)airbyte/sources/util.py (1)
get_source(47-142)
airbyte/_executors/java.py (6)
airbyte/_executors/base.py (1)
Executor(159-248)airbyte/_util/telemetry.py (2)
EventState(168-172)log_install_state(298-311)airbyte/version.py (1)
get_version(12-14)airbyte/sources/registry.py (1)
ConnectorMetadata(57-88)airbyte/_message_iterators.py (1)
AirbyteMessageIterator(61-205)airbyte/exceptions.py (2)
AirbyteConnectorInstallationError(344-345)AirbyteSubprocessFailedError(274-277)
🪛 GitHub Actions: Run Linters
examples/run_source_snowflake_java.py
[error] 96-96: Command: poetry run mypy .; mypy error: examples/run_source_snowflake_java.py:96: Statement is unreachable [unreachable].
airbyte/_executors/java.py
[error] 247-247: Command: poetry run mypy .; mypy error: airbyte/_executors/java.py:247: Item "None" of "IO[bytes] | None" has no attribute "enter" [union-attr].
[error] 247-247: Command: poetry run mypy .; mypy error: airbyte/_executors/java.py:247: Item "None" of "IO[bytes] | None" has no attribute "exit" [union-attr].
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (Fast)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (No Creds)
airbyte/_executors/java.py
Outdated
| for member in tar: | ||
| if root_dir is None: | ||
| root_dir = member.name.split("/")[0] | ||
|
|
||
| if member.name.startswith(root_dir + "/"): | ||
| stripped_name = member.name[len(root_dir) + 1 :] | ||
| if not stripped_name: # Skip empty names (root directory itself) | ||
| continue | ||
|
|
||
| target_path = self.jre_dir / stripped_name | ||
|
|
||
| if member.isdir(): | ||
| target_path.mkdir(parents=True, exist_ok=True) | ||
| elif member.isfile(): | ||
| target_path.parent.mkdir(parents=True, exist_ok=True) | ||
| with tar.extractfile(member) as source: | ||
| if source: | ||
| target_path.write_bytes(source.read()) | ||
| target_path.chmod(member.mode) | ||
| elif member.issym(): | ||
| target_path.parent.mkdir(parents=True, exist_ok=True) | ||
| target_path.symlink_to(member.linkname) | ||
| elif member.name == root_dir: |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add path traversal guards in JRE strip-components extraction
Even though Azul archives are trusted, a quick guard prevents surprises. Would you reject absolute paths and “..” segments in stripped_name, wdyt?
- if member.name.startswith(root_dir + "/"):
+ if member.name.startswith(root_dir + "/"):
stripped_name = member.name[len(root_dir) + 1 :]
if not stripped_name: # Skip empty names (root directory itself)
continue
- target_path = self.jre_dir / stripped_name
+ # Safety: disallow absolute paths and parent traversal
+ if stripped_name.startswith("/") or ".." in Path(stripped_name).parts:
+ raise exc.AirbyteConnectorInstallationError(
+ message="JRE archive contains unsafe path.",
+ connector_name=self.name,
+ context={"unsafe_path": stripped_name},
+ )
+ target_path = self.jre_dir / stripped_name📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for member in tar: | |
| if root_dir is None: | |
| root_dir = member.name.split("/")[0] | |
| if member.name.startswith(root_dir + "/"): | |
| stripped_name = member.name[len(root_dir) + 1 :] | |
| if not stripped_name: # Skip empty names (root directory itself) | |
| continue | |
| target_path = self.jre_dir / stripped_name | |
| if member.isdir(): | |
| target_path.mkdir(parents=True, exist_ok=True) | |
| elif member.isfile(): | |
| target_path.parent.mkdir(parents=True, exist_ok=True) | |
| with tar.extractfile(member) as source: | |
| if source: | |
| target_path.write_bytes(source.read()) | |
| target_path.chmod(member.mode) | |
| elif member.issym(): | |
| target_path.parent.mkdir(parents=True, exist_ok=True) | |
| target_path.symlink_to(member.linkname) | |
| elif member.name == root_dir: | |
| for member in tar: | |
| if root_dir is None: | |
| root_dir = member.name.split("/")[0] | |
| if member.name.startswith(root_dir + "/"): | |
| stripped_name = member.name[len(root_dir) + 1 :] | |
| if not stripped_name: # Skip empty names (root directory itself) | |
| continue | |
| # Safety: disallow absolute paths and parent traversal | |
| if stripped_name.startswith("/") or ".." in Path(stripped_name).parts: | |
| raise exc.AirbyteConnectorInstallationError( | |
| message="JRE archive contains unsafe path.", | |
| connector_name=self.name, | |
| context={"unsafe_path": stripped_name}, | |
| ) | |
| target_path = self.jre_dir / stripped_name | |
| if member.isdir(): | |
| target_path.mkdir(parents=True, exist_ok=True) | |
| elif member.isfile(): | |
| target_path.parent.mkdir(parents=True, exist_ok=True) | |
| with tar.extractfile(member) as source: | |
| if source: | |
| target_path.write_bytes(source.read()) | |
| target_path.chmod(member.mode) | |
| elif member.issym(): | |
| target_path.parent.mkdir(parents=True, exist_ok=True) | |
| target_path.symlink_to(member.linkname) | |
| elif member.name == root_dir: | |
| # (rest of loop unchanged) |
🧰 Tools
🪛 GitHub Actions: Run Linters
[error] 247-247: Command: poetry run mypy .; mypy error: airbyte/_executors/java.py:247: Item "None" of "IO[bytes] | None" has no attribute "enter" [union-attr].
[error] 247-247: Command: poetry run mypy .; mypy error: airbyte/_executors/java.py:247: Item "None" of "IO[bytes] | None" has no attribute "exit" [union-attr].
🤖 Prompt for AI Agents
In airbyte/_executors/java.py around lines 232-254, the extraction logic should
guard against path traversal by rejecting stripped_name values that are absolute
or contain parent ("..") segments; before creating target_path, check
Path(stripped_name).is_absolute() or '..' in Path(stripped_name).parts and skip
the member (optionally log a warning) if true. Apply the same guard for symlink
targets: if member.issym() and Path(member.linkname).is_absolute() or '..' in
Path(member.linkname).parts then skip creating the symlink. Ensure these checks
happen immediately after computing stripped_name so no filesystem writes or
chmod/symlink creation occur for unsafe paths.
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (6)
examples/run_java_connector_tests.py (6)
4-6: Fix the Poetry usage command in the docstring.Running a script file via Poetry usually requires invoking python; suggesting this small tweak to reduce friction, wdyt?
- poetry run examples/run_java_connector_tests.py + poetry run python examples/run_java_connector_tests.py
24-31: Simplify unsetting JAVA_HOME.You can compress this logic by using pop(), keeping the same behavior and messages, wdyt?
def unset_java_home() -> None: """Unset JAVA_HOME environment variable to avoid conflicts with auto-downloaded JRE.""" - if "JAVA_HOME" in os.environ: - del os.environ["JAVA_HOME"] + if os.environ.pop("JAVA_HOME", None) is not None: print("✅ Unset JAVA_HOME to avoid conflicts with auto-downloaded JRE.") else: print("ℹ️ JAVA_HOME was not set, no need to unset it.")
37-53: Add timeouts and status checks to Google Drive downloads.A couple of small hardening tweaks will help avoid writing an HTML error page to a .tar and make transient issues clearer while keeping the example simple: add a timeout and raise_for_status() on both requests, and close the session via a context manager, wdyt?
- # Create session and get initial response - session = requests.Session() - response = session.get(f"https://drive.google.com/uc?export=download&id={file_id}") + # Create session and get initial response + with requests.Session() as session: + response = session.get( + f"https://drive.google.com/uc?export=download&id={file_id}", + timeout=60, + ) + response.raise_for_status() - if "virus scan warning" in response.text.lower(): - uuid_match = re.search(r'name="uuid" value="([^"]+)"', response.text) - uuid_value = uuid_match.group(1) if uuid_match else "" + if "virus scan warning" in response.text.lower(): + uuid_match = re.search(r'name="uuid" value="([^"]+)"', response.text) + uuid_value = uuid_match.group(1) if uuid_match else "" - response = session.get( - f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}" - ) + response = session.get( + f"https://drive.usercontent.google.com/download?id={file_id}&export=download&confirm=t&uuid={uuid_value}", + timeout=60, + ) + response.raise_for_status() fd, temp_path = tempfile.mkstemp(suffix=".tar") temp_file = Path(temp_path) os.close(fd) temp_file.write_bytes(response.content) return temp_file
14-18: Prepare for a small refinement below (imports).If we tweak the record slicing to avoid materializing the full generator and broaden a return type, we’ll need these imports, wdyt?
import os import re import tempfile from pathlib import Path + +from itertools import islice +from typing import Any
56-60: Broaden the return type for connector config.Connector configs often include non-string values (booleans, numbers). Widening the type better reflects reality and avoids misleading hints, wdyt?
def get_connector_config( connector_name: str, secret_name: str | None = None, -) -> dict[str, str]: +) -> dict[str, Any]:
106-114: Avoid materializing the full stream and guard against empty discovery.Using list(... )[:10] will read the entire generator before slicing; switching to itertools.islice keeps it at 10 records. Also, a quick guard avoids an IndexError if no streams are discovered. These are small, user-friendly tweaks for the demo, wdyt?
stream_names = source.get_available_streams() print(f"📊 Found {len(stream_names)} streams") + if not stream_names: + print("⚠️ No streams discovered; exiting.") + return + selected_stream = stream_names[0] source.select_streams([selected_stream]) + print(f"🔎 Reading first 10 records from stream: {selected_stream}") read_result = source.read() - records = list(read_result[selected_stream])[:10] + records = list(islice(read_result[selected_stream], 10)) print(f"✅ Read {len(records)} records using Java connector!")
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
examples/run_java_connector_tests.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (1)
examples/*
⚙️ CodeRabbit Configuration File
These scripts are intended for demonstration purposes only. They are not meant to represent production code. For these scripts, you should prefer brevity over robust error handling and handling of edge cases. These are demos - which should be as simple as possible to handle the 'blessed' use cases.
Files:
examples/run_java_connector_tests.py
🧬 Code Graph Analysis (1)
examples/run_java_connector_tests.py (2)
airbyte/secrets/google_gsm.py (4)
GoogleGSMSecretManager(75-298)GSMSecretHandle(54-72)fetch_connector_secret(264-298)get_secret_handle(172-190)airbyte/sources/base.py (2)
get_available_streams(321-323)select_streams(243-272)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (No Creds)
- GitHub Check: Pytest (Fast)
🔇 Additional comments (3)
examples/run_java_connector_tests.py (3)
1-3: Nice, clear demo entry point and docstring.This example is concise and does a great job walking a user through the Java path, including JRE conflicts and config retrieval. 👍
61-65: Optional: upfront env check for credentials.Given this is an example, brevity is good. Still, a tiny guard can make failures friendlier if the env var is missing. Would you like to add a short assertion or message before constructing the secret manager, e.g., in main()?
- Example:
- if "DEVIN_GCP_SERVICE_ACCOUNT_JSON" not in os.environ: print("Set DEVIN_GCP_SERVICE_ACCOUNT_JSON"); return
92-99: Confirm that Path for use_java_tar and implicit use_java are supported.Per the executor factory updates, passing only use_java_tar should imply Java mode and accept a Path-like. Can you confirm that:
- get_source accepts Path for use_java_tar (not just str)?
- setting use_java_tar alone is sufficient to trigger Java execution?
If either is not guaranteed, we might explicitly cast to str and/or set use_java=True for clarity, wdyt?
- Add None check before using tar.extractfile() in context manager - Resolves union-attr errors in streaming tar extraction - Maintains functionality while satisfying type checker Co-Authored-By: AJ Steers <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (9)
airbyte/_executors/java.py (9)
28-28: ** Shadowing built-in print function**This duplicates a previous review comment from Copilot about shadowing the built-in
70-90: ** Auto-detect TAR path not implemented**This duplicates a previous review comment about
_auto_detect_tar_pathreturningNone, which breaks Docker→Java fallback when no explicituse_java_taris provided.
79-80: ** Cache directory should include Java version**This duplicates a previous review comment about the JRE cache path missing the Java version, which can cause stale JRE reuse.
127-142: ** Print statements should use logging**This duplicates a previous review comment from Copilot about using consistent logging mechanisms instead of direct print statements.
232-254: ** Add path traversal guards in JRE extraction**This duplicates a previous review comment about adding safety guards for path traversal in the JRE extraction logic.
247-250: ** Fix mypy error with extractfile**This duplicates a previous review comment about the mypy error where
tar.extractfile()can returnNoneand doesn't implement the context manager protocol.
273-274: ** Security: Consider validating tar contents before extraction**This duplicates a previous review comment about adding validation before using
tar.extractall()to prevent path traversal attacks.
298-304: ** Executable discovery needs tightening**This duplicates a previous review comment about improving the executable discovery logic to prefer name matches, avoid symlinks, and ensure paths remain within the connector directory.
387-390: ** Hardcoded command mapping could be more flexible**This duplicates a previous review comment from Copilot about making the argument mapping logic more flexible instead of hardcoding specific command values.
🧹 Nitpick comments (3)
airbyte/_executors/java.py (3)
409-418: Consider handling stdin writing in a separate thread for robustnessWriting to stdin sequentially could block if the Java process's stdin buffer fills up before it starts consuming, potentially causing deadlock. Would you consider using a separate thread for stdin writing to prevent blocking, wdyt?
+import threading + if isinstance(stdin, AirbyteMessageIterator): - try: - for message in stdin: - if process.stdin: - process.stdin.write(message.model_dump_json() + "\n") - process.stdin.flush() - if process.stdin: - process.stdin.close() - except BrokenPipeError: - pass + def write_stdin(): + try: + for message in stdin: + if process.stdin: + process.stdin.write(message.model_dump_json() + "\n") + process.stdin.flush() + if process.stdin: + process.stdin.close() + except BrokenPipeError: + pass + + stdin_thread = threading.Thread(target=write_stdin) + stdin_thread.start()
432-432: Exit code -15 handling needs clarificationThe code treats exit code
-15(SIGTERM) as success alongside0. Could you add a comment explaining why-15is considered successful? Is this for graceful shutdowns during connector interruption, wdyt?- if exit_code not in {0, -15}: + # Exit codes: 0 = success, -15 = SIGTERM (graceful shutdown) + if exit_code not in {0, -15}:
420-428: Potential resource leak with threading approachIf we implement the threading suggestion above, we should ensure the stdin thread is properly joined to prevent resource leaks. Also, the current stdout reading could block if the process hangs. Would you consider adding a timeout or ensuring thread cleanup, wdyt?
+ stdin_thread = None if isinstance(stdin, AirbyteMessageIterator): # ... threading code from above suggestion ... + stdin_thread = threading.Thread(target=write_stdin) + stdin_thread.start() if process.stdout: try: while True: line = process.stdout.readline() if not line: break yield line.rstrip("\n\r") finally: process.stdout.close() + # Ensure stdin thread completes + if stdin_thread: + stdin_thread.join(timeout=30) # Reasonable timeout + exit_code = process.wait()
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
airbyte/_executors/java.py(1 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (1)
airbyte/_executors/java.py (6)
airbyte/_executors/base.py (1)
Executor(159-248)airbyte/_util/telemetry.py (2)
EventState(168-172)log_install_state(298-311)airbyte/version.py (1)
get_version(12-14)airbyte/sources/registry.py (1)
ConnectorMetadata(57-88)airbyte/_message_iterators.py (1)
AirbyteMessageIterator(61-205)airbyte/exceptions.py (2)
AirbyteConnectorInstallationError(344-345)AirbyteSubprocessFailedError(274-277)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (6)
- GitHub Check: Pytest (All, Python 3.11, Windows)
- GitHub Check: Pytest (All, Python 3.11, Ubuntu)
- GitHub Check: Pytest (No Creds)
- GitHub Check: Pytest (All, Python 3.10, Windows)
- GitHub Check: Pytest (All, Python 3.10, Ubuntu)
- GitHub Check: Pytest (Fast)
Summary
This PR implements Java connector support for PyAirbyte by copying and enhancing the implementation from PR #719. The key addition is a dual-parameter API that provides fine-grained control over Java connector execution:
use_java: Controls Java execution mode (None/True/False/Pathto JRE)use_java_tar: Specifies connector TAR file location (Nonefor auto-detect,Pathfor specific file)The implementation includes automatic JRE management, connector TAR extraction, and intelligent fallback logic (Docker when available, Java when not). A complete example script demonstrates usage with source-snowflake, including Google Drive TAR download and credential handling patterns.
Core changes:
JavaExecutorclass with automatic JRE download from Azul APIget_source()API to include both Java parametersexamples/run_source_snowflake_java.pydemonstrating real-world usageReview & Testing Checklist for Human
use_javaanduse_java_tarparameters to ensure the interaction logic works as expected and documentedRecommended test plan: Run the example script with real Snowflake test credentials, then try variations with different parameter combinations and intentional failure scenarios.
Diagram
%%{ init : { "theme" : "default" }}%% graph TD A["airbyte/sources/util.py<br/>get_source()"]:::minor-edit --> B["airbyte/_executors/util.py<br/>get_connector_executor()"]:::minor-edit B --> C["airbyte/_executors/java.py<br/>JavaExecutor"]:::major-edit D["examples/run_source_snowflake_java.py<br/>Demo script"]:::major-edit --> A C --> E["JRE Download<br/>Azul API"]:::context C --> F["TAR Extraction<br/>~/.airbyte/connectors/"]:::context C --> G["Java Process<br/>Connector execution"]:::context subgraph Legend L1[Major Edit]:::major-edit L2[Minor Edit]:::minor-edit L3[Context/No Edit]:::context end classDef major-edit fill:#90EE90 classDef minor-edit fill:#87CEEB classDef context fill:#FFFFFFNotes
Link to Devin run: https://app.devin.ai/sessions/e9a8bcdfcab246f0857ac38f3755296f
Requested by: @aaronsteers
Summary by CodeRabbit
New Features
Documentation