arunkumarucet commented on code in PR #19434:
URL: https://github.com/apache/pinot/pull/19434#discussion_r3957710549
##########
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);
+ try {
+ return Files.readAllBytes(localFile.toPath());
+ } finally {
+ Files.deleteIfExists(localFile.toPath());
+ Files.deleteIfExists(localFile.getParentFile().toPath());
Review Comment:
Fixed in 55a40be by removing the local-copy lifecycle from the descriptor
path entirely: the content is now streamed via `PinotFS.open(URI)` with
try-with-resources — no temp dir, no sidecar interaction.
`getFileCopiedToLocal` (still used for the codegen jar) now recursively deletes
its temp tree when the copy throws, so partial content and checksum sidecars
are removed too. Coverage:
`testGetFileCopiedToLocalCleansUpTempDirOnCopyFailure` simulates a Hadoop-style
copy that writes partial content plus a `.crc` sidecar before throwing and
asserts the directory is fully removed, and
`testFetchesFreshOnEveryCallAndNeverCopiesToLocal` asserts the descriptor path
never enters the copy lifecycle.
##########
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));
Review Comment:
Fixed in 55a40be: promotion now requires the actual decoder-level resolution
to succeed. The new `ProtoBufUtils.getDescriptor(path, messageTypeName,
fallback)` parses the set and resolves the configured `protoClassName` (or
requires a non-empty set and takes the first message type when unset), and only
then publishes to the cache — so an empty descriptor set fails with "contains
no message types" and can never displace a good copy. Regression test as
requested: `testUnresolvableFetchedContentFailsAndPreservesFallbackCopy` warms
a good entry, serves `new byte[0]`, asserts the call fails and the good
fallback remains intact (also covered for corrupt bytes and a missing message
type).
--
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]
