This is an automated email from the ASF dual-hosted git repository.
KKcorps pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 15b838a5df8 Fall back to last-known-good protobuf descriptor when the
remote fetch fails (#19434)
15b838a5df8 is described below
commit 15b838a5df8d1cc34ab6302289defed31c7f44f9
Author: Arunkumar Saravanan <[email protected]>
AuthorDate: Thu Sep 17 11:25:32 2026 +0530
Fall back to last-known-good protobuf descriptor when the remote fetch
fails (#19434)
A remote (e.g. S3) descriptor file is re-fetched on every decoder
creation, i.e. on every CONSUMING transition on every replica, making the
transition a hard dependency on live DNS + object store. A transient DNS
blip at rollover time permanently marks consuming segments ERROR.
The descriptor is still fetched fresh on every call (streamed via
PinotFS.open, building on #19378), so in-place updates keep propagating,
but the last remote content that both fetched and resolved successfully
is now remembered per URI and served only when the fetch itself fails:
- Content is promoted to the cache only after the configured message type
resolves (ProtoBufUtils.getDescriptor), so an empty or unusable
descriptor set can never displace a good copy; this also removes the
decoder's second parse of the descriptor.
- Only fetch (I/O) failures fall back; fetched-but-unresolvable content
fails decoder init with a distinct error and leaves the cache intact.
- Each fetch is stamped with its start time and published via merge, so a
slow stale fetch cannot roll the cache back over a newer entry.
- Cache entries weigh content + key + fixed overhead, bounding both
memory and entry count.
- Controllable via cluster config
pinot.server.protobuf.descriptor.fallback.enabled (dynamically
updatable, default true), overridden per table by the decoder prop
descriptorFileFallbackEnabled.
- getFileCopiedToLocal (codegen jar path) now removes its temp tree on
copy failure.
---
.../protobuf/ProtoBufMessageDecoder.java | 39 ++-
.../plugin/inputformat/protobuf/ProtoBufUtils.java | 135 ++++++++-
.../protobuf/ProtoBufMessageDecoderTest.java | 25 ++
.../protobuf/ProtoBufUtilsDescriptorCacheTest.java | 312 +++++++++++++++++++++
.../server/starter/helix/BaseServerStarter.java | 5 +
.../apache/pinot/spi/utils/CommonConstants.java | 8 +
.../utils/ProtoBufDescriptorFallbackListener.java | 87 ++++++
.../ProtoBufDescriptorFallbackListenerTest.java | 75 +++++
8 files changed, 660 insertions(+), 26 deletions(-)
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
index 5c79564f2ac..197c46006da 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
@@ -18,27 +18,30 @@
*/
package org.apache.pinot.plugin.inputformat.protobuf;
-import com.github.os72.protobuf.dynamic.DynamicSchema;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.Message;
-import java.io.IOException;
-import java.io.InputStream;
import java.util.Map;
import java.util.Set;
-import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.stream.StreamMessageDecoder;
+import org.apache.pinot.spi.utils.ProtoBufDescriptorFallbackListener;
//TODO: Add support for Schema Registry
public class ProtoBufMessageDecoder implements StreamMessageDecoder<byte[]> {
public static final String DESCRIPTOR_FILE_PATH = "descriptorFile";
public static final String PROTO_CLASS_NAME = "protoClassName";
+ /// When enabled, a remote descriptor that both fetched and resolved
successfully before is served when the
+ /// remote fetch fails, so a transient DNS / object-store outage does not
permanently fail the CONSUMING
+ /// transition. When this decoder prop is not set, the cluster config
+ /// 'pinot.server.protobuf.descriptor.fallback.enabled' applies (enabled by
default, dynamically updatable);
+ /// setting the prop to 'true' or 'false' overrides the cluster-wide value
for the table.
+ public static final String DESCRIPTOR_FILE_FALLBACK_ENABLED =
"descriptorFileFallbackEnabled";
private ProtoBufRecordExtractor _recordExtractor;
- private String _protoClassName;
private Message.Builder _builder;
@Override
@@ -47,30 +50,20 @@ public class ProtoBufMessageDecoder implements
StreamMessageDecoder<byte[]> {
Preconditions.checkState(props.containsKey(DESCRIPTOR_FILE_PATH),
"Protocol Buffer schema descriptor file must be provided");
- _protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
- Descriptors.Descriptor descriptor;
- try (InputStream fin =
ProtoBufUtils.openDescriptorFile(props.get(DESCRIPTOR_FILE_PATH))) {
- descriptor = buildProtoBufDescriptor(fin);
- }
+ Descriptors.Descriptor descriptor =
ProtoBufUtils.getDescriptor(props.get(DESCRIPTOR_FILE_PATH),
+ props.getOrDefault(PROTO_CLASS_NAME, ""),
isDescriptorFallbackEnabled(props));
_recordExtractor = new ProtoBufRecordExtractor();
_recordExtractor.init(fieldsToRead, null);
DynamicMessage dynamicMessage =
DynamicMessage.getDefaultInstance(descriptor);
_builder = dynamicMessage.newBuilderForType();
}
- private Descriptors.Descriptor buildProtoBufDescriptor(InputStream fin)
- throws IOException {
- try {
- DynamicSchema dynamicSchema = DynamicSchema.parseFrom(fin);
-
- if (!StringUtils.isEmpty(_protoClassName)) {
- return dynamicSchema.getMessageDescriptor(_protoClassName);
- } else {
- return
dynamicSchema.getMessageDescriptor(dynamicSchema.getMessageTypes().toArray(new
String[]{})[0]);
- }
- } catch (Descriptors.DescriptorValidationException e) {
- throw new IOException("Descriptor file validation failed", e);
- }
+ /// The table-level decoder prop, when set, overrides the dynamically
updatable cluster-wide setting.
+ @VisibleForTesting
+ static boolean isDescriptorFallbackEnabled(Map<String, String> props) {
+ String tableOverride = props.get(DESCRIPTOR_FILE_FALLBACK_ENABLED);
+ return tableOverride != null ? Boolean.parseBoolean(tableOverride)
+ : ProtoBufDescriptorFallbackListener.getInstance().isEnabled();
}
@Override
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
index 9ec6110054c..a91b397051c 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtils.java
@@ -18,6 +18,11 @@
*/
package org.apache.pinot.plugin.inputformat.protobuf;
+import com.github.os72.protobuf.dynamic.DynamicSchema;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
import com.google.protobuf.Descriptors;
import com.google.protobuf.ProtobufInternalUtils;
import java.io.File;
@@ -25,6 +30,11 @@ import java.io.InputStream;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.Set;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.filesystem.PinotFSFactory;
import org.slf4j.Logger;
@@ -35,12 +45,41 @@ public class ProtoBufUtils {
public static final String TMP_DIR_PREFIX = "pinot-protobuf";
public static final String PB_OUTER_CLASS_SUFFIX = "OuterClass";
+ // Last content of each remote (S3, GCS, ...) descriptor file that both
fetched and resolved successfully, 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
itself 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. The weight charges content, key and a
fixed per-entry overhead so both total
+ // memory and entry count stay bounded.
+ private static final long FALLBACK_CACHE_MAX_WEIGHT_BYTES = 64L << 20;
+ private static final int FALLBACK_CACHE_ENTRY_OVERHEAD_BYTES = 1024;
+ private static final Cache<String, CachedDescriptor>
LAST_KNOWN_GOOD_DESCRIPTORS = CacheBuilder.newBuilder()
+ .maximumWeight(FALLBACK_CACHE_MAX_WEIGHT_BYTES)
+ .weigher((String key, CachedDescriptor value) ->
+ value._content.length + 2 * key.length() +
FALLBACK_CACHE_ENTRY_OVERHEAD_BYTES)
+ .build();
+
+ /// Descriptor content stamped with the time its fetch started. Publication
keeps the entry whose fetch started
+ /// last, so a slow stale fetch that completes after a newer one can never
roll the cache backward.
+ private static class CachedDescriptor {
+ final byte[] _content;
+ final long _fetchStartNanos;
+
+ CachedDescriptor(byte[] content, long fetchStartNanos) {
+ _content = content;
+ _fetchStartNanos = fetchStartNanos;
+ }
+ }
+
private ProtoBufUtils() {
}
/// Copies a remote file to a local temp directory and returns the local
copy. Used by the codegen
/// decoder path for JAR files that must remain on disk for the lifetime of
the classloader.
- /// The caller is responsible for managing the lifetime of the returned file
and its parent directory.
+ /// On success the caller is responsible for managing the lifetime of the
returned file and its parent directory;
+ /// on copy failure the temp directory is removed (the copy may have left
partial content or filesystem sidecars
+ /// such as Hadoop `.crc` files). This is a plain download with no outage
fallback (a jar cannot be validated as
+ /// a descriptor set); descriptors should be resolved through
[#getDescriptor(String, String, boolean)] instead.
public static File getFileCopiedToLocal(String filePath)
throws Exception {
URI fileURI = URI.create(filePath);
@@ -54,8 +93,13 @@ public class ProtoBufUtils {
File localFile = createLocalFile(fileURI, localTmpDir.toFile());
LOGGER.info("Copying protocol buffer jar/descriptor file from source: {}
to dst: {}", filePath,
localFile.getAbsolutePath());
- pinotFS.copyToLocalFile(fileURI, localFile);
- return localFile;
+ try {
+ pinotFS.copyToLocalFile(fileURI, localFile);
+ return localFile;
+ } catch (Exception e) {
+ deleteRecursivelyQuietly(localTmpDir);
+ throw e;
+ }
} else {
throw new RuntimeException(String.format("Scheme: %s not supported in
PinotFSFactory"
+ " for protocol buffer jar/descriptor file: %s.", scheme,
filePath));
@@ -86,6 +130,91 @@ public class ProtoBufUtils {
return pinotFS.open(fileURI);
}
+ /// Resolves a message [Descriptors.Descriptor] from the descriptor set at
the given path: the message type with
+ /// the given name, or the first message type in the set when the name is
null or empty.
+ ///
+ /// The descriptor set is read fresh on every call via
[#openDescriptorFile(String)] — no temp files — so
+ /// in-place updates of the file keep propagating. When
`fallbackToLastKnownGood` is set, the last remote content
+ /// that both fetched and resolved successfully is remembered per URI and
served ONLY when the fetch itself fails
+ /// (e.g. a transient DNS or object-store outage), so decoder creation
survives outages instead of permanently
+ /// marking the CONSUMING segment ERROR. Content that fetches successfully
but does not resolve (corrupt, empty,
+ /// or missing the requested message type) always fails the call and leaves
the remembered copy untouched: a bad
+ /// descriptor deployment must surface as an error, never silently serve an
obsolete schema. Local files are
+ /// read fresh and never remembered.
+ ///
+ /// NOTE: Only this method has the fallback. The jar used by
[ProtoBufCodeGenMessageDecoder] is downloaded via
+ /// [#getFileCopiedToLocal(String)] without one (see the note there).
+ public static Descriptors.Descriptor getDescriptor(String
descriptorFilePath, @Nullable String messageTypeName,
+ boolean fallbackToLastKnownGood)
+ throws Exception {
+ String scheme = URI.create(descriptorFilePath).getScheme();
+ boolean remote = scheme != null &&
!scheme.equals(PinotFSFactory.LOCAL_PINOT_FS_SCHEME);
+ boolean fallbackEnabled = remote && fallbackToLastKnownGood;
+ // The stamp is taken before the fetch so that publication can reject a
slow stale fetch that completes after
+ // a newer one (see CachedDescriptor)
+ long fetchStartNanos = System.nanoTime();
+ byte[] content;
+ try (InputStream inputStream = openDescriptorFile(descriptorFilePath)) {
+ content = inputStream.readAllBytes();
+ } catch (Exception fetchException) {
+ CachedDescriptor lastKnownGood =
+ fallbackEnabled ?
LAST_KNOWN_GOOD_DESCRIPTORS.getIfPresent(descriptorFilePath) : null;
+ if (lastKnownGood == null) {
+ throw fetchException;
+ }
+ LOGGER.warn("Failed to fetch protocol buffer descriptor file: {},
falling back to the last known good copy",
+ descriptorFilePath, fetchException);
+ return resolveMessageDescriptor(lastKnownGood._content, messageTypeName,
descriptorFilePath);
+ }
+ // A fetched-but-unresolvable descriptor fails here, before publication,
so it can neither be served nor
+ // overwrite the last known good copy
+ Descriptors.Descriptor descriptor = resolveMessageDescriptor(content,
messageTypeName, descriptorFilePath);
+ if (fallbackEnabled) {
+ LAST_KNOWN_GOOD_DESCRIPTORS.asMap().merge(descriptorFilePath,
+ new CachedDescriptor(content, fetchStartNanos),
+ (existing, candidate) -> candidate._fetchStartNanos -
existing._fetchStartNanos > 0 ? candidate
+ : existing);
+ }
+ return descriptor;
+ }
+
+ /// Parses the descriptor set and resolves the requested message type (or
the first one when no name is given).
+ /// Failures here mean the content is unusable — deliberately distinct from
a fetch failure.
+ private static Descriptors.Descriptor resolveMessageDescriptor(byte[]
descriptorSetBytes,
+ @Nullable String messageTypeName, String descriptorFilePath)
+ throws Exception {
+ DynamicSchema schema;
+ try {
+ schema = DynamicSchema.parseFrom(descriptorSetBytes);
+ } catch (Exception e) {
+ throw new IllegalStateException("Invalid protocol buffer descriptor set
at: " + descriptorFilePath, e);
+ }
+ String typeName = messageTypeName;
+ if (StringUtils.isEmpty(typeName)) {
+ Set<String> messageTypes = schema.getMessageTypes();
+ Preconditions.checkState(!messageTypes.isEmpty(), "Descriptor set at: %s
contains no message types",
+ descriptorFilePath);
+ typeName = messageTypes.iterator().next();
+ }
+ Descriptors.Descriptor descriptor = schema.getMessageDescriptor(typeName);
+ Preconditions.checkState(descriptor != null, "Message type: %s not found
in descriptor set at: %s", typeName,
+ descriptorFilePath);
+ return descriptor;
+ }
+
+ private static void deleteRecursivelyQuietly(Path dir) {
+ try (Stream<Path> paths = Files.walk(dir)) {
+ paths.sorted(Comparator.reverseOrder()).forEach(path ->
path.toFile().delete());
+ } catch (Exception e) {
+ LOGGER.warn("Failed to clean up temporary directory: {}", dir, e);
+ }
+ }
+
+ @VisibleForTesting
+ static void clearDescriptorCache() {
+ LAST_KNOWN_GOOD_DESCRIPTORS.invalidateAll();
+ }
+
public static String getFullJavaName(Descriptors.Descriptor descriptor) {
String prefix;
if (null != descriptor.getContainingType()) {
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoderTest.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoderTest.java
index 20511192689..a8f93b98be9 100644
---
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoderTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoderTest.java
@@ -25,15 +25,40 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.ProtoBufDescriptorFallbackListener;
import org.testng.annotations.Test;
import static
org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.*;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
public class ProtoBufMessageDecoderTest {
+ @Test
+ public void testDescriptorFallbackEnabledPrecedence() {
+ ProtoBufDescriptorFallbackListener listener =
ProtoBufDescriptorFallbackListener.getInstance();
+ try {
+ Map<String, String> props = new HashMap<>();
+ // Without a table-level prop, the (dynamically updatable) cluster-wide
value applies
+ listener.setEnabled(true);
+ assertTrue(ProtoBufMessageDecoder.isDescriptorFallbackEnabled(props));
+ listener.setEnabled(false);
+ assertFalse(ProtoBufMessageDecoder.isDescriptorFallbackEnabled(props));
+
+ // An explicit table-level prop overrides the cluster-wide value in both
directions
+ props.put(ProtoBufMessageDecoder.DESCRIPTOR_FILE_FALLBACK_ENABLED,
"true");
+ assertTrue(ProtoBufMessageDecoder.isDescriptorFallbackEnabled(props));
+ listener.setEnabled(true);
+ props.put(ProtoBufMessageDecoder.DESCRIPTOR_FILE_FALLBACK_ENABLED,
"false");
+ assertFalse(ProtoBufMessageDecoder.isDescriptorFallbackEnabled(props));
+ } finally {
+ listener.reset();
+ }
+ }
+
@Test
public void testHappyCase()
throws Exception {
diff --git
a/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtilsDescriptorCacheTest.java
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtilsDescriptorCacheTest.java
new file mode 100644
index 00000000000..4ad5e91fcdb
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-protobuf/src/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufUtilsDescriptorCacheTest.java
@@ -0,0 +1,312 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.protobuf;
+
+import com.github.os72.protobuf.dynamic.DynamicSchema;
+import java.io.File;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.pinot.spi.filesystem.LocalPinotFS;
+import org.apache.pinot.spi.filesystem.PinotFSFactory;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Tests the last-known-good fallback in [ProtoBufUtils#getDescriptor(String,
String, boolean)]: the descriptor is
+/// fetched fresh on every call (so in-place updates propagate) and the last
content that both fetched and resolved
+/// successfully is served only when the fetch itself fails.
Fetched-but-unresolvable content (corrupt, empty, or
+/// missing the requested type) must fail the call without touching the
remembered copy, and a slow stale fetch must
+/// never roll the remembered copy backward.
+public class ProtoBufUtilsDescriptorCacheTest {
+ private static final String COUNTING_SCHEME = "counting";
+
+ private File _descriptorFile;
+ private byte[] _sampleContent;
+ private String _sampleTypeName;
+ private byte[] _complexContent;
+ private String _complexTypeName;
+
+ /// A "remote" filesystem that serves local files, counts reads, and can
simulate network failures, stalled
+ /// streams, and copies that leave partial content plus checksum sidecars
behind (like Hadoop filesystems do).
+ public static class CountingPinotFS extends LocalPinotFS {
+ static final AtomicInteger OPEN_CALLS = new AtomicInteger();
+ static volatile boolean _failReads = false;
+ static volatile CountDownLatch _firstOpenStallLatch = null;
+ static final CountDownLatch[] FIRST_OPEN_REACHED_EOF = new
CountDownLatch[1];
+ static final AtomicReference<File> LAST_COPY_DST = new AtomicReference<>();
+
+ @Override
+ public InputStream open(URI uri)
+ throws IOException {
+ int call = OPEN_CALLS.incrementAndGet();
+ if (_failReads) {
+ throw new IOException("Simulated network failure: Temporary failure in
name resolution");
+ }
+ InputStream delegate = super.open(uri);
+ CountDownLatch stallLatch = _firstOpenStallLatch;
+ if (stallLatch != null && call == 1) {
+ return stallAtEof(delegate, stallLatch);
+ }
+ return delegate;
+ }
+
+ @Override
+ public void copyToLocalFile(URI srcUri, File dstFile)
+ throws Exception {
+ LAST_COPY_DST.set(dstFile);
+ if (_failReads) {
+ // Simulate a Hadoop-style filesystem failing mid-copy after writing
partial content and a checksum
+ // sidecar into the destination directory
+ Files.write(dstFile.toPath(), new byte[]{1, 2, 3});
+ Files.write(new File(dstFile.getParentFile(), "." + dstFile.getName()
+ ".crc").toPath(), new byte[]{4});
+ throw new IOException("Simulated network failure: Temporary failure in
name resolution");
+ }
+ super.copyToLocalFile(srcUri, dstFile);
+ }
+
+ /// Wraps the stream so that the first reader announces reaching EOF and
then blocks until released — used to
+ /// hold a stale fetch open while a newer fetch completes.
+ private static InputStream stallAtEof(InputStream delegate, CountDownLatch
stallLatch) {
+ return new FilterInputStream(delegate) {
+ @Override
+ public int read()
+ throws IOException {
+ int result = super.read();
+ if (result == -1) {
+ awaitRelease();
+ }
+ return result;
+ }
+
+ @Override
+ public int read(byte[] buffer, int offset, int length)
+ throws IOException {
+ int result = super.read(buffer, offset, length);
+ if (result == -1) {
+ awaitRelease();
+ }
+ return result;
+ }
+
+ private void awaitRelease()
+ throws IOException {
+ FIRST_OPEN_REACHED_EOF[0].countDown();
+ try {
+ stallLatch.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ };
+ }
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ PinotFSFactory.register(COUNTING_SCHEME, CountingPinotFS.class.getName(),
null);
+ _descriptorFile = File.createTempFile("proto-descriptor-cache-test",
".desc");
+ try (InputStream in =
getClass().getClassLoader().getResourceAsStream("sample.desc")) {
+ _sampleContent = in.readAllBytes();
+ }
+ try (InputStream in =
getClass().getClassLoader().getResourceAsStream("complex_types.desc")) {
+ _complexContent = in.readAllBytes();
+ }
+ _sampleTypeName = firstMessageType(_sampleContent);
+ _complexTypeName = firstMessageType(_complexContent);
+ }
+
+ @AfterClass
+ public void tearDown() {
+ _descriptorFile.delete();
+ }
+
+ @BeforeMethod
+ public void reset()
+ throws Exception {
+ ProtoBufUtils.clearDescriptorCache();
+ CountingPinotFS.OPEN_CALLS.set(0);
+ CountingPinotFS._failReads = false;
+ CountingPinotFS._firstOpenStallLatch = null;
+ CountingPinotFS.FIRST_OPEN_REACHED_EOF[0] = new CountDownLatch(1);
+ CountingPinotFS.LAST_COPY_DST.set(null);
+ Files.write(_descriptorFile.toPath(), _sampleContent);
+ }
+
+ private String remotePath() {
+ return COUNTING_SCHEME + "://" + _descriptorFile.getAbsolutePath();
+ }
+
+ private static String firstMessageType(byte[] descriptorSetBytes)
+ throws Exception {
+ return
DynamicSchema.parseFrom(descriptorSetBytes).getMessageTypes().iterator().next();
+ }
+
+ @Test
+ public void testFetchesFreshOnEveryCallAndNeverCopiesToLocal()
+ throws Exception {
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+ assertEquals(CountingPinotFS.OPEN_CALLS.get(), 1);
+
+ // An in-place update of the remote file must be picked up by the next
fetch
+ Files.write(_descriptorFile.toPath(), _complexContent);
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _complexTypeName);
+ assertEquals(CountingPinotFS.OPEN_CALLS.get(), 2);
+
+ // The fetch path streams the content and never goes through the
local-copy (temp file) lifecycle
+ assertNull(CountingPinotFS.LAST_COPY_DST.get());
+ }
+
+ @Test
+ public void testFallsBackToLastKnownGoodOnFetchFailure()
+ throws Exception {
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+
+ CountingPinotFS._failReads = true;
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+ // The fetch was attempted (and failed) before falling back
+ assertEquals(CountingPinotFS.OPEN_CALLS.get(), 2);
+
+ // Recovery: once the network is back, fresh content is fetched and
remembered again
+ CountingPinotFS._failReads = false;
+ Files.write(_descriptorFile.toPath(), _complexContent);
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _complexTypeName);
+ }
+
+ @Test
+ public void testFetchFailureWithoutFallbackCopyPropagates() {
+ CountingPinotFS._failReads = true;
+ assertThrows(IOException.class, () ->
ProtoBufUtils.getDescriptor(remotePath(), null, true));
+ }
+
+ @Test
+ public void testFallbackDisabledFailsFastEvenWhenWarm()
+ throws Exception {
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+ CountingPinotFS._failReads = true;
+ assertThrows(IOException.class, () ->
ProtoBufUtils.getDescriptor(remotePath(), null, false));
+ }
+
+ @Test
+ public void testUnresolvableFetchedContentFailsAndPreservesFallbackCopy()
+ throws Exception {
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+
+ // An empty descriptor set parses but resolves no message type: it must
fail the call, not serve the fallback
+ Files.write(_descriptorFile.toPath(), new byte[0]);
+ IllegalStateException emptyFailure =
+ expectThrows(IllegalStateException.class, () ->
ProtoBufUtils.getDescriptor(remotePath(), null, true));
+ assertTrue(emptyFailure.getMessage().contains("no message types"),
emptyFailure.getMessage());
+
+ // Unparseable bytes (0xFF = invalid wire type) must fail the same way
+ Files.write(_descriptorFile.toPath(), new byte[]{(byte) 0xFF, (byte) 0xFF,
(byte) 0xFF, (byte) 0xFF});
+ IllegalStateException corruptFailure =
+ expectThrows(IllegalStateException.class, () ->
ProtoBufUtils.getDescriptor(remotePath(), null, true));
+ assertTrue(corruptFailure.getMessage().contains("Invalid protocol buffer
descriptor set"),
+ corruptFailure.getMessage());
+
+ // A valid set missing the requested message type must also fail without
promotion
+ Files.write(_descriptorFile.toPath(), _complexContent);
+ assertThrows(IllegalStateException.class,
+ () -> ProtoBufUtils.getDescriptor(remotePath(), "does.not.Exist",
true));
+
+ // None of the failures above may have replaced the last known good copy
+ CountingPinotFS._failReads = true;
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _sampleTypeName);
+ }
+
+ @Test(timeOut = 60_000)
+ public void testStaleConcurrentFetchCannotRollFallbackCopyBackward()
+ throws Exception {
+ // Thread A fetches the OLD content and stalls at end-of-stream, before
publication
+ CountDownLatch stallLatch = new CountDownLatch(1);
+ CountingPinotFS._firstOpenStallLatch = stallLatch;
+ AtomicReference<Throwable> staleFetchFailure = new AtomicReference<>();
+ Thread staleFetch = new Thread(() -> {
+ try {
+ ProtoBufUtils.getDescriptor(remotePath(), null, true);
+ } catch (Throwable t) {
+ staleFetchFailure.set(t);
+ }
+ });
+ staleFetch.start();
+ CountingPinotFS.FIRST_OPEN_REACHED_EOF[0].await();
+
+ // While A is stalled, the remote file is replaced and a newer fetch
completes and publishes
+ Files.write(_descriptorFile.toPath(), _complexContent);
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _complexTypeName);
+
+ // A resumes and attempts to publish the older content; publication must
reject it
+ stallLatch.countDown();
+ staleFetch.join();
+ assertNull(staleFetchFailure.get());
+
+ CountingPinotFS._failReads = true;
+ assertEquals(ProtoBufUtils.getDescriptor(remotePath(), null,
true).getFullName(), _complexTypeName);
+ }
+
+ @Test
+ public void testLocalDescriptorReadFreshWithoutFallback()
+ throws Exception {
+ File localFile = File.createTempFile("proto-descriptor-cache-test-local",
".desc");
+ try {
+ Files.write(localFile.toPath(), _sampleContent);
+ assertEquals(ProtoBufUtils.getDescriptor(localFile.getAbsolutePath(),
null, true).getFullName(),
+ _sampleTypeName);
+
+ // A local file edited in place must be re-read
+ Files.write(localFile.toPath(), _complexContent);
+ assertEquals(ProtoBufUtils.getDescriptor(localFile.getAbsolutePath(),
null, true).getFullName(),
+ _complexTypeName);
+
+ // Local files are never remembered: once the file is gone the call
fails even though it succeeded before
+ assertTrue(localFile.delete());
+ assertThrows(Exception.class, () ->
ProtoBufUtils.getDescriptor(localFile.getAbsolutePath(), null, true));
+ } finally {
+ localFile.delete();
+ }
+ }
+
+ @Test
+ public void testGetFileCopiedToLocalCleansUpTempDirOnCopyFailure() {
+ CountingPinotFS._failReads = true;
+ assertThrows(IOException.class, () ->
ProtoBufUtils.getFileCopiedToLocal(remotePath()));
+
+ // The temp directory must be fully removed even though the failed copy
left partial content and a checksum
+ // sidecar behind
+ File copyDst = CountingPinotFS.LAST_COPY_DST.get();
+ assertFalse(copyDst.getParentFile().exists(), "Leaked temp directory: " +
copyDst.getParentFile());
+ }
+}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index d43a54b0c64..a2d64283a9e 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -140,6 +140,7 @@ import
org.apache.pinot.spi.utils.ConsumingSegmentConsistencyModeListener;
import org.apache.pinot.spi.utils.InstanceTypeUtils;
import org.apache.pinot.spi.utils.NetUtils;
import org.apache.pinot.spi.utils.PinotMd5Mode;
+import org.apache.pinot.spi.utils.ProtoBufDescriptorFallbackListener;
import org.apache.pinot.spi.utils.builder.TableNameBuilder;
import org.apache.pinot.sql.parsers.rewriter.QueryRewriterFactory;
import org.slf4j.Logger;
@@ -284,6 +285,10 @@ public abstract class BaseServerStarter implements
ServiceStartable {
ConsumingSegmentConsistencyModeListener.getInstance());
LOGGER.info(
"Registered ConsumingSegmentConsistencyModeListener change listener
for dynamic force commit/reload control");
+ // Register configuration change listener for the protobuf descriptor
fallback setting
+
_clusterConfigChangeHandler.registerClusterConfigChangeListener(ProtoBufDescriptorFallbackListener.getInstance());
+ LOGGER.info("Registered ProtoBufDescriptorFallbackListener change listener
for dynamic descriptor fallback"
+ + " control");
LOGGER.info("Initializing Helix manager with zkAddress: {}, clusterName:
{}, instanceId: {}", _zkAddress,
_helixClusterName, _instanceId);
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index e0de1210a65..a7314eee62e 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -3004,5 +3004,13 @@ public class CommonConstants {
/// - PROTECTED: Force commit is enabled with metadata reversion on
inconsistencies
/// - UNSAFE: Force commit is enabled without metadata reversion (Can lead
to inconsistencies)
public static final String CONSUMING_SEGMENT_CONSISTENCY_MODE =
"pinot.server.consuming.segment.consistency.mode";
+
+ /// Cluster config key to control whether the protobuf decoder falls back
to the last successfully fetched
+ /// (and resolved) descriptor when the remote descriptor fetch fails, so a
transient DNS / object-store outage
+ /// does not permanently fail the CONSUMING transition. Enabled by
default; set to 'false' to fail fast
+ /// instead. Dynamically updatable without a server restart; a table-level
decoder prop
+ /// ('descriptorFileFallbackEnabled') overrides this cluster-wide value.
+ public static final String PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED =
+ "pinot.server.protobuf.descriptor.fallback.enabled";
}
}
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListener.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListener.java
new file mode 100644
index 00000000000..46b2859328a
--- /dev/null
+++
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListener.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.utils;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
+import
org.apache.pinot.spi.utils.CommonConstants.ConfigChangeListenerConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Singleton holding the cluster-wide default for the protobuf decoder's
descriptor fallback: whether the last
+/// successfully fetched (and resolved) descriptor is served when the remote
descriptor fetch fails, so a transient
+/// DNS / object-store outage does not permanently fail the CONSUMING
transition. Enabled by default.
+///
+/// Dynamically updatable via the ZK cluster config
+/// [ConfigChangeListenerConstants#PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED]
without a server restart; the value is
+/// consulted each time a decoder is created. A table-level decoder prop
('descriptorFileFallbackEnabled')
+/// overrides this cluster-wide value. Thread-safe.
+public class ProtoBufDescriptorFallbackListener implements
PinotClusterConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ProtoBufDescriptorFallbackListener.class);
+ private static final boolean DEFAULT_ENABLED = true;
+ private static final ProtoBufDescriptorFallbackListener INSTANCE = new
ProtoBufDescriptorFallbackListener();
+
+ private final AtomicBoolean _enabled = new AtomicBoolean(DEFAULT_ENABLED);
+
+ private ProtoBufDescriptorFallbackListener() {
+ }
+
+ public static ProtoBufDescriptorFallbackListener getInstance() {
+ return INSTANCE;
+ }
+
+ public boolean isEnabled() {
+ return _enabled.get();
+ }
+
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ if
(!changedConfigs.contains(ConfigChangeListenerConstants.PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED))
{
+ return;
+ }
+ boolean newEnabled =
+
fromString(clusterConfigs.get(ConfigChangeListenerConstants.PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED));
+ boolean previousEnabled = _enabled.getAndSet(newEnabled);
+ if (previousEnabled != newEnabled) {
+ LOGGER.info("Updated cluster config: {} from {} to {}",
+ ConfigChangeListenerConstants.PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED,
previousEnabled, newEnabled);
+ }
+ }
+
+ private static boolean fromString(String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return DEFAULT_ENABLED;
+ }
+ return Boolean.parseBoolean(value.trim());
+ }
+
+ @VisibleForTesting
+ public void reset() {
+ _enabled.set(DEFAULT_ENABLED);
+ }
+
+ @VisibleForTesting
+ public void setEnabled(boolean enabled) {
+ _enabled.set(enabled);
+ }
+}
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListenerTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListenerTest.java
new file mode 100644
index 00000000000..92c784bc90d
--- /dev/null
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/ProtoBufDescriptorFallbackListenerTest.java
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.spi.utils;
+
+import java.util.Map;
+import java.util.Set;
+import
org.apache.pinot.spi.utils.CommonConstants.ConfigChangeListenerConstants;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+public class ProtoBufDescriptorFallbackListenerTest {
+ private static final String KEY =
ConfigChangeListenerConstants.PROTOBUF_DESCRIPTOR_FALLBACK_ENABLED;
+
+ private final ProtoBufDescriptorFallbackListener _listener =
ProtoBufDescriptorFallbackListener.getInstance();
+
+ @AfterMethod
+ public void reset() {
+ _listener.reset();
+ }
+
+ @Test
+ public void testEnabledByDefault() {
+ assertTrue(_listener.isEnabled());
+ }
+
+ @Test
+ public void testDisableAndReEnable() {
+ _listener.onChange(Set.of(KEY), Map.of(KEY, "false"));
+ assertFalse(_listener.isEnabled());
+ _listener.onChange(Set.of(KEY), Map.of(KEY, "true"));
+ assertTrue(_listener.isEnabled());
+ }
+
+ @Test
+ public void testRemovedOrBlankValueRestoresDefault() {
+ _listener.onChange(Set.of(KEY), Map.of(KEY, "false"));
+ assertFalse(_listener.isEnabled());
+ // Key removed from the cluster config
+ _listener.onChange(Set.of(KEY), Map.of());
+ assertTrue(_listener.isEnabled());
+
+ _listener.onChange(Set.of(KEY), Map.of(KEY, "false"));
+ assertFalse(_listener.isEnabled());
+ _listener.onChange(Set.of(KEY), Map.of(KEY, " "));
+ assertTrue(_listener.isEnabled());
+ }
+
+ @Test
+ public void testUnrelatedChangeIgnored() {
+ _listener.onChange(Set.of(KEY), Map.of(KEY, "false"));
+ assertFalse(_listener.isEnabled());
+ _listener.onChange(Set.of("some.other.config"),
Map.of("some.other.config", "value", KEY, "true"));
+ assertFalse(_listener.isEnabled());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
