arunkumarucet commented on code in PR #19434:
URL: https://github.com/apache/pinot/pull/19434#discussion_r3957714337
##########
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:
Fixed in 55a40be: fetch and resolution are now separate stages, and only a
fetch (I/O) exception can trigger the fallback. Content that fetches
successfully but does not resolve — corrupt, empty, or missing the configured
message type — fails decoder init with a distinct
`IllegalStateException("Invalid protocol buffer descriptor set at: ...")` (vs.
the fetch-failure WARN) and leaves the remembered copy untouched, so a bad
descriptor deployment surfaces immediately instead of silently serving the
previous schema. Covered by
`testUnresolvableFetchedContentFailsAndPreservesFallbackCopy`.
##########
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:
Fixed in 55a40be: the weigher now charges `content.length + 2 * key.length()
+ 1024` fixed per-entry overhead, so the 64MB budget bounds both total memory
(including key/entry bookkeeping) and the entry count (~65K max). Zero-weight
entries are additionally impossible since promotion now requires the configured
message type to resolve, so an empty descriptor set is never cached.
--
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]
