arunkumarucet commented on code in PR #19434:
URL: https://github.com/apache/pinot/pull/19434#discussion_r3957712464
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java:
##########
@@ -60,9 +81,53 @@ public static File getFileCopiedToLocal(String filePath)
}
}
+ /// Returns the content of the descriptor file at the given path. A remote
file is fetched fresh on every call so
+ /// in-place updates are picked up, and the last successfully fetched (and
parseable) content is remembered per
+ /// URI: when the fetch fails, or returns bytes that do not parse as a
descriptor set, the remembered copy is
+ /// served instead so that decoder creation survives transient DNS /
object-store outages. Local files are always
+ /// read fresh and never remembered.
+ ///
+ /// NOTE: Only descriptor files get this fallback. The jar used by
[ProtoBufCodeGenMessageDecoder] is downloaded
+ /// via [#getFileCopiedToLocal(String)] without one (see the note there).
public static InputStream getDescriptorFileInputStream(String
descriptorFilePath)
throws Exception {
- return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
+ URI fileURI = URI.create(descriptorFilePath);
+ String scheme = fileURI.getScheme();
+ if (scheme == null || scheme.equals(PinotFSFactory.LOCAL_PINOT_FS_SCHEME))
{
+ return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
+ }
+ byte[] content;
+ try {
+ content = downloadFileToBytes(descriptorFilePath);
+ // Validate before remembering so that a corrupt/truncated download can
neither be served nor overwrite the
+ // last known good copy
+ DynamicSchema.parseFrom(new ByteArrayInputStream(content));
+ LAST_KNOWN_GOOD_DESCRIPTORS.put(descriptorFilePath, content);
+ } catch (Exception e) {
+ content = LAST_KNOWN_GOOD_DESCRIPTORS.getIfPresent(descriptorFilePath);
+ if (content == null) {
+ throw e;
+ }
+ LOGGER.warn("Failed to fetch protocol buffer descriptor file: {},
falling back to the last successfully"
+ + " fetched copy", descriptorFilePath, e);
+ }
+ return new ByteArrayInputStream(content);
+ }
+
+ private static byte[] downloadFileToBytes(String filePath)
+ throws Exception {
+ File localFile = getFileCopiedToLocal(filePath);
Review Comment:
Fixed in 55a40be: the descriptor fetch path no longer creates temp
directories at all (streamed via `PinotFS.open`), so there is nothing to leak
on a fetch failure. For the remaining local-copy user (`getFileCopiedToLocal`,
codegen jar), temp creation, copy, and cleanup are now under one owner: the
copy is wrapped and the whole temp tree is recursively deleted on failure.
Cleanup on the failure path is asserted in
`testGetFileCopiedToLocalCleansUpTempDirOnCopyFailure`.
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java:
##########
@@ -60,9 +81,53 @@ public static File getFileCopiedToLocal(String filePath)
}
}
+ /// Returns the content of the descriptor file at the given path. A remote
file is fetched fresh on every call so
+ /// in-place updates are picked up, and the last successfully fetched (and
parseable) content is remembered per
+ /// URI: when the fetch fails, or returns bytes that do not parse as a
descriptor set, the remembered copy is
+ /// served instead so that decoder creation survives transient DNS /
object-store outages. Local files are always
+ /// read fresh and never remembered.
+ ///
+ /// NOTE: Only descriptor files get this fallback. The jar used by
[ProtoBufCodeGenMessageDecoder] is downloaded
+ /// via [#getFileCopiedToLocal(String)] without one (see the note there).
public static InputStream getDescriptorFileInputStream(String
descriptorFilePath)
throws Exception {
- return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
+ URI fileURI = URI.create(descriptorFilePath);
+ String scheme = fileURI.getScheme();
+ if (scheme == null || scheme.equals(PinotFSFactory.LOCAL_PINOT_FS_SCHEME))
{
+ return new FileInputStream(getFileCopiedToLocal(descriptorFilePath));
+ }
+ byte[] content;
+ try {
+ content = downloadFileToBytes(descriptorFilePath);
+ // Validate before remembering so that a corrupt/truncated download can
neither be served nor overwrite the
+ // last known good copy
+ DynamicSchema.parseFrom(new ByteArrayInputStream(content));
+ LAST_KNOWN_GOOD_DESCRIPTORS.put(descriptorFilePath, content);
Review Comment:
Fixed in 55a40be with a generation-aware publication rule: each fetch is
stamped with its start time (`System.nanoTime()` taken before the read), and
publication goes through `asMap().merge` keeping the entry whose fetch started
latest — so a stale fetch that completes after a newer one is rejected rather
than rolling the cache backward. Deterministic concurrent regression test as
requested: `testStaleConcurrentFetchCannotRollFallbackCopyBackward` stalls the
first reader at end-of-stream via a latch, lets a newer fetch of replaced
content complete and publish, releases the stale reader, and asserts a
subsequent outage serves the newer content.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
