KKcorps commented on code in PR #19434:
URL: https://github.com/apache/pinot/pull/19434#discussion_r3956615196
##########
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:
**[Critical] Cleanup breaks descriptor loads through Hadoop-backed
filesystems.** With Hadoop 3.4.3, `copyToLocalFile()` writes a hidden checksum
sidecar (for example, `.descriptor.desc.crc`) beside the local descriptor. This
block deletes only `localFile`, then attempts to delete its still-nonempty
parent, which throws `DirectoryNotEmptyException`. I reproduced that exception
here, so a cold HDFS load fails; if the cache is warm, every successful refresh
instead falls back to stale bytes. Please avoid the local-copy lifecycle for
descriptor reads—`PinotFS.open(URI)` can be consumed with try-with-resources—or
ensure the entire temp tree is safely owned and removed. Please add coverage
using a checksum-producing filesystem.
##########
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:
**[Major] This is not sufficient validation for cache promotion.**
`DynamicSchema.parseFrom()` accepts an empty descriptor set; I verified that
`new byte[0]` produces a schema with no message types. The following `put` can
therefore replace a valid cached descriptor with unusable bytes, after which
the caller fails while resolving its configured message type. Please promote
bytes only after the actual decoder validates the required descriptor/message
(or supply that validation through a callback). A regression test should warm a
good entry, return an empty descriptor, and verify that the good fallback
remains intact.
##########
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:
**[Major] Concurrent successful refreshes can roll this cache backward.**
There is no ordering between fetch completion and this unconditional `put`:
request A can fetch old bytes and pause, request B can fetch and cache a newer
descriptor, and then A can overwrite it with the old value. A later outage
serves the older schema. A deterministic probe reproduced `old fetch -> new
fetch -> old fallback`. Please serialize refreshes per URI or use a
generation/version-aware publication rule, with a concurrent regression test.
##########
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) {
Review Comment:
**[Major] Please do not treat descriptor validation failures as transient
fetch outages.** This catch covers both remote I/O and parsing. If the remote
object is successfully replaced with a malformed or incompatible descriptor,
silently serving the previous schema can make the decoder interpret new
payloads with obsolete metadata instead of surfacing the bad deployment.
Preserve the cached copy, but restrict fallback to genuine I/O failures; a
fetched-but-invalid descriptor should fail the current initialization.
##########
pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java:
##########
@@ -36,9 +41,25 @@ public class ProtoBufUtils {
public static final String TMP_DIR_PREFIX = "pinot-protobuf";
public static final String PB_OUTER_CLASS_SUFFIX = "OuterClass";
+ // Last successfully fetched (and parseable) content of each remote (S3,
GCS, ...) descriptor file, keyed by URI.
+ // The descriptor is still fetched fresh on every decoder creation, so
in-place updates of the file keep
+ // propagating exactly as before; this copy is served only when the fetch
fails (e.g. a transient DNS or
+ // object-store outage), so a CONSUMING transition cannot go to ERROR on a
network blip once the descriptor has
+ // been fetched once by this JVM. Bounded by total content size as a safety
net.
+ private static final long FALLBACK_CACHE_MAX_WEIGHT_BYTES = 64L << 20;
+ private static final Cache<String, byte[]> LAST_KNOWN_GOOD_DESCRIPTORS =
CacheBuilder.newBuilder()
+ .maximumWeight(FALLBACK_CACHE_MAX_WEIGHT_BYTES)
+ .weigher((String key, byte[] value) -> value.length)
Review Comment:
**[Major] The 64 MiB limit does not bound total cache memory.** This weighs
only `value.length`, excluding the URI string, entry/node overhead, and array
overhead. More importantly, currently accepted empty descriptor sets have
weight zero, so unique remote URIs can create an unbounded number of entries
without consuming the configured weight. Please require semantically usable
descriptors, charge a positive per-entry/key overhead, and/or add a maximum
entry count.
##########
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:
**[Major] The fetch-failure path leaks its temporary directory.**
`getFileCopiedToLocal()` creates the temp directory before calling
`copyToLocalFile()`. When that copy throws, this method never receives a
`File`, so execution cannot enter the `finally` below. I reproduced a leftover
`pinot-protobuf*` directory after the simulated network failure—the exact path
this fallback is intended to handle. Please keep temp creation, copy, and
cleanup under one owner/finally, or read through `PinotFS.open(URI)`, and
assert cleanup in the failure test.
--
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]
