[Backport to master] Analytics core migration - #1782
Conversation
* Add flag for analytics-write path * Updated flag name * Rename setAnalyticsWriteEnabled to setAnalyticsCoreWriteEnabled
…am (#1767) * Integrate Analytics core write path to connector * Updated code for Gemini Code Assist comments * Updated code to wrap Analytics core output stream in BufferedOutputStream * Fixed buidl failure * Updated code for the review comments * Fixed excetion handling with getFileInfo * Added unit tests for improving code coverage * Address review comments for Analytics Core write path integration * Add a TODO for exception handling * Updated as per review comments
* Map hadoop-connector write flags to gcs-analytics core flags * Fix flag mapping for checksum * Updated code as per review comments * Updated code for review comments
* Add integration tests * Refactor setupGhfs to accept URI and improve exception assertions in integration tests * Fixed test formatting * standardize setupGhfs and add byte slice write test * Add a TODO and updated tests to not use hard-coded values * Update tests for the comments * Update TODO
|
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request integrates the GCS Analytics Core library into the GCS Hadoop Connector. It provides the necessary infrastructure to map existing Hadoop configurations to the new library, wraps input and output streams to support the new implementation, and adds feature flags to control the migration. Extensive testing has been included to ensure compatibility and performance. Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request integrates the GCS Analytics Core library into the GCS Hadoop Connector, introducing configuration properties to enable it, mapping configurations, and wrapping the input and output streams to route operations through the new library. Feedback on the changes highlights several critical areas for improvement: addressing thread-safety and idempotency issues in the input stream wrapper's close() method; removing redundant synchronized modifiers in the output stream wrapper to adhere to the style guide (and updating tests accordingly); preventing a potential NullPointerException in vectored reads; ensuring proper input validation in readFully(); avoiding eager initialization of the GCS filesystem to preserve lazy initialization; and replacing fragile string-matching exception handling with proper exception translation.
| public void close() throws IOException { | ||
| if (channelIsOpen) { | ||
| inputStream.close(); | ||
| channelIsOpen = false; | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
The channelIsOpen flag is not declared as volatile, and the close() method is not synchronized. Since readVectored is not synchronized and can be executed concurrently by multiple threads, there is a visibility and thread-safety issue. If close() is called concurrently with other operations, or called multiple times, it can lead to race conditions or multiple close calls on the underlying stream. Making channelIsOpen volatile and synchronizing close() ensures thread safety and idempotency.
| public void close() throws IOException { | |
| if (channelIsOpen) { | |
| inputStream.close(); | |
| channelIsOpen = false; | |
| } | |
| } | |
| @Override | |
| public synchronized void close() throws IOException { | |
| if (channelIsOpen) { | |
| channelIsOpen = false; | |
| inputStream.close(); | |
| } | |
| } |
| @Override | ||
| public synchronized void write(int b) throws IOException { | ||
| delegate.write(b); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void write(byte[] b, int off, int len) throws IOException { | ||
| logger.atFine().log("write(byte[], off=%d, len=%d)", off, len); | ||
| delegate.write(b, off, len); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void close() throws IOException { | ||
| if (closed) { | ||
| logger.atFiner().log("close(): Stream already closed, ignoring."); | ||
| return; | ||
| } | ||
| try { | ||
| delegate.close(); | ||
| } finally { | ||
| closed = true; | ||
| } | ||
| } |
There was a problem hiding this comment.
According to the Repository Style Guide (line 31), redundant synchronized modifiers should not be added to wrapper classes or methods where the underlying stream (such as Hadoop's FSDataOutputStream) is single-threaded by specification. Since this output stream wrapper is always used within Hadoop's FSDataOutputStream write path, the synchronized modifiers on write and close methods are redundant and should be removed.
@Override
public void write(int b) throws IOException {
delegate.write(b);
}
@Override
public void write(byte[] b, int off, int len) throws IOException {
logger.atFine().log("write(byte[], off=%d, len=%d)", off, len);
delegate.write(b, off, len);
}
@Override
public void close() throws IOException {
if (closed) {
logger.atFiner().log("close(): Stream already closed, ignoring.");
return;
}
try {
delegate.close();
} finally {
closed = true;
}
}References
- Do NOT add redundant synchronized modifiers to wrapper classes or methods where the underlying stream (such as Hadoop's FSDataOutputStream) is single-threaded by specification. (link)
| @Test | ||
| public void methodsAreSynchronized() throws NoSuchMethodException { | ||
| verifyMethodIsSynchronized(GcsAnalyticsCoreOutputStreamWrapper.class, "write", int.class); | ||
| verifyMethodIsSynchronized( | ||
| GcsAnalyticsCoreOutputStreamWrapper.class, "write", byte[].class, int.class, int.class); | ||
| verifyMethodIsSynchronized(GcsAnalyticsCoreOutputStreamWrapper.class, "close"); | ||
| } |
There was a problem hiding this comment.
If the redundant synchronized modifiers are removed from GcsAnalyticsCoreOutputStreamWrapper to adhere to the style guide, the methodsAreSynchronized test will fail. This test should be removed or updated accordingly.
| CompletableFuture<?>[] futures = | ||
| ranges.stream().map(FileRange::getData).toArray(CompletableFuture[]::new); |
There was a problem hiding this comment.
If any range's data future is not initialized or is set to null (e.g., due to an unexpected issue in the underlying readVectored implementation), CompletableFuture.allOf(futures) will throw a NullPointerException. Adding a defensive filter to exclude null futures is highly recommended.
CompletableFuture<?>[] futures =
ranges.stream()
.map(FileRange::getData)
.filter(java.util.Objects::nonNull)
.toArray(CompletableFuture[]::new);| public synchronized void readFully(long position, byte[] buffer, int offset, int length) | ||
| throws IOException { | ||
| if (length == 0) { | ||
| return; | ||
| } | ||
| if (channel instanceof GcsAnalyticsCoreInputStreamWrapper) { |
There was a problem hiding this comment.
When channel is an instance of GcsAnalyticsCoreInputStreamWrapper, the readFully method bypasses super.readFully and directly delegates to the wrapper. However, this also bypasses the standard input validation performed by super.readFully (such as checking if position < 0). To ensure consistent fail-fast behavior and compatibility with the Hadoop specification, an explicit check for position < 0 should be added.
@Override
public synchronized void readFully(long position, byte[] buffer, int offset, int length)
throws IOException {
if (length == 0) {
return;
}
if (position < 0) {
throw new java.io.EOFException("Explicit read position cannot be negative: " + position);
}| // TODO(user): Initialize analyticsCoreGcsFs lazily when GCS_LAZY_INITIALIZATION_ENABLE is | ||
| // true, to avoid eager initialization. | ||
| if (isAnalyticsCoreEnabled() || isAnalyticsCoreWriteEnabled()) { | ||
| analyticsCoreGcsFs = createAnalyticsGcsFs(config); | ||
| } |
There was a problem hiding this comment.
Eagerly initializing analyticsCoreGcsFs during initialize() defeats the purpose of lazy initialization when GCS_LAZY_INITIALIZATION_ENABLE is set to true. This can cause unnecessary credential resolution and connection setup on client/driver nodes where GCS is not actually accessed. analyticsCoreGcsFs should be initialized lazily on demand (e.g., inside getAnalyticsCoreGcsFs()).
| } catch (IOException e) { | ||
| // TODO(user): Remove this exception message check once analytics-core throws a specific | ||
| // FileNotFoundException / NoSuchFileException for non-existent objects. | ||
| if (e.getMessage() != null && e.getMessage().startsWith("Object not found:")) { | ||
| // If file does not exist, set the generation as 0. | ||
| generation = 0L; | ||
| } else { | ||
| throw e; | ||
| } | ||
| } |
There was a problem hiding this comment.
Relying on e.getMessage().startsWith("Object not found:") to detect a non-existent file is extremely fragile and prone to breaking if the underlying library changes its error message format or localization. According to the style guide (line 43), lower-level exceptions should be translated into standard Hadoop IOException subclasses (such as FileNotFoundException). The analytics-core library or the connector's adapter layer should throw FileNotFoundException for non-existent objects, allowing this block to catch FileNotFoundException directly.
References
- Lower-level GCS/gRPC exceptions must be translated into standard Hadoop IOException subclasses. (link)
7 participants
No description provided.