You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/03/03 03:29:40 UTC
(pinot) branch security/pinot-avro-remove-md5-wording-clean created (now 56f01c262b7)
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a change to branch security/pinot-avro-remove-md5-wording-clean
in repository https://gitbox.apache.org/repos/asf/pinot.git
at 56f01c262b7 Refactor schema hash naming in KafkaAvroMessageDecoder.
This branch includes the following new commits:
new 56f01c262b7 Refactor schema hash naming in KafkaAvroMessageDecoder.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/01: Refactor schema hash naming in KafkaAvroMessageDecoder.
Posted by xi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xiangfu pushed a commit to branch security/pinot-avro-remove-md5-wording-clean
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 56f01c262b764647236dc43623b07f71ce12c1be
Author: Xiang Fu <xi...@startree.ai>
AuthorDate: Mon Mar 2 17:21:58 2026 -0800
Refactor schema hash naming in KafkaAvroMessageDecoder.
This removes MD5-specific wording from internal identifiers and comments in the pinot-avro decoder to align terminology with FIPS-compliant hash usage without changing behavior.
Made-with: Cursor
---
.../inputformat/avro/KafkaAvroMessageDecoder.java | 70 +++++++++++-----------
1 file changed, 35 insertions(+), 35 deletions(-)
diff --git a/pinot-plugins/pinot-input-format/pinot-avro/src/main/java/org/apache/pinot/plugin/inputformat/avro/KafkaAvroMessageDecoder.java b/pinot-plugins/pinot-input-format/pinot-avro/src/main/java/org/apache/pinot/plugin/inputformat/avro/KafkaAvroMessageDecoder.java
index c21dbe2356a..3d1f39afc40 100644
--- a/pinot-plugins/pinot-input-format/pinot-avro/src/main/java/org/apache/pinot/plugin/inputformat/avro/KafkaAvroMessageDecoder.java
+++ b/pinot-plugins/pinot-input-format/pinot-avro/src/main/java/org/apache/pinot/plugin/inputformat/avro/KafkaAvroMessageDecoder.java
@@ -60,15 +60,15 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
private static final String SCHEMA_REGISTRY_REST_URL = "schema.registry.rest.url";
private static final String SCHEMA_REGISTRY_SCHEMA_NAME = "schema.registry.schema.name";
private org.apache.avro.Schema _defaultAvroSchema;
- private MD5AvroSchemaMap _md5ToAvroSchemaMap;
+ private HashToSchemaMap _hashToSchemaMap;
// A global cache for schemas across all threads.
private static final Map<String, org.apache.avro.Schema> GLOBAL_SCHEMA_CACHE = new HashMap<>();
// Suffix for getting the latest schema
private static final String LATEST = "-latest";
- // Reusable byte[] to read MD5 from payload. This is OK as this class is used only by a single thread.
- private final byte[] _reusableMD5Bytes = new byte[SCHEMA_HASH_LENGTH];
+ // Reusable byte[] to read hash from payload. This class is used only by a single thread.
+ private final byte[] _reusableHashBytes = new byte[SCHEMA_HASH_LENGTH];
private DecoderFactory _decoderFactory;
private RecordExtractor<GenericData.Record> _avroRecordExtractor;
@@ -98,9 +98,9 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
// With the logic below, we may not set defaultAvroSchema to be the latest one everytime.
// The schema is fetched once when the machine starts. Until the next restart. the latest schema is
// not fetched.
- // But then we always pay attention to the exact MD5 hash and attempt to fetch the schema for that particular hash
- // before decoding an incoming kafka event. We use defaultAvroSchema only if the fetch for the particular MD5 fails,
- // but then we will retry that fetch on every event in case of failure.
+ // But then we always pay attention to the exact hash and attempt to fetch the schema for that particular hash
+ // before decoding an incoming kafka event. We use defaultAvroSchema only if the fetch for the particular hash
+ // fails, but then we will retry that fetch on every event in case of failure.
synchronized (GLOBAL_SCHEMA_CACHE) {
final String hashKey = avroSchemaName + LATEST;
_defaultAvroSchema = GLOBAL_SCHEMA_CACHE.get(hashKey);
@@ -125,7 +125,7 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
_avroRecordExtractor = PluginManager.get().createInstance(recordExtractorClass);
_avroRecordExtractor.init(fieldsToRead, config);
_decoderFactory = new DecoderFactory();
- _md5ToAvroSchemaMap = new MD5AvroSchemaMap();
+ _hashToSchemaMap = new HashToSchemaMap();
}
@Nullable
@@ -142,25 +142,25 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
return null;
}
- System.arraycopy(payload, SCHEMA_HASH_START_OFFSET + offset, _reusableMD5Bytes, 0, SCHEMA_HASH_LENGTH);
+ System.arraycopy(payload, SCHEMA_HASH_START_OFFSET + offset, _reusableHashBytes, 0, SCHEMA_HASH_LENGTH);
boolean schemaUpdateFailed = false;
- org.apache.avro.Schema schema = _md5ToAvroSchemaMap.getSchema(_reusableMD5Bytes);
+ org.apache.avro.Schema schema = _hashToSchemaMap.getSchema(_reusableHashBytes);
if (schema == null) {
// We will get here for the first row consumed in the segment, and every row that has a schema ID that is
- // not yet in md5ToAvroSchemaMap.
+ // not yet in hashToAvroSchemaMap.
synchronized (GLOBAL_SCHEMA_CACHE) {
- final String hashKey = hex(_reusableMD5Bytes);
+ final String hashKey = hex(_reusableHashBytes);
schema = GLOBAL_SCHEMA_CACHE.get(hashKey);
if (schema == null) {
// We will get here only if no partition of the table has populated the global schema cache.
// In that case, one of the consumers will fetch the schema and populate the cache, and the others
// should find it in the cache and po
- final String schemaUri = "/id=" + hex(_reusableMD5Bytes);
+ final String schemaUri = "/id=" + hex(_reusableHashBytes);
try {
schema = fetchSchema(schemaUri);
GLOBAL_SCHEMA_CACHE.put(hashKey, schema);
- _md5ToAvroSchemaMap.addSchema(_reusableMD5Bytes, schema);
+ _hashToSchemaMap.addSchema(_reusableHashBytes, schema);
} catch (Exception e) {
schema = _defaultAvroSchema;
LOGGER.error("Error fetching schema using url {}. Attempting to continue with previous schema", schemaUri,
@@ -169,7 +169,7 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
}
} else {
LOGGER.info("Found schema for {} in cache", hashKey);
- _md5ToAvroSchemaMap.addSchema(_reusableMD5Bytes, schema);
+ _hashToSchemaMap.addSchema(_reusableHashBytes, schema);
}
}
}
@@ -199,8 +199,8 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
private static class SchemaFetcher implements Callable<Boolean> {
private org.apache.avro.Schema _schema;
- private URL _url;
- private boolean _isSuccessful = false;
+ private final URL _url;
+ private final boolean _isSuccessful = false;
SchemaFetcher(URL url) {
_url = url;
@@ -248,33 +248,33 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
}
/**
- * Private class for encapsulating MD5 to Avro schema mapping.
+ * Private class for encapsulating hash to Avro schema mapping.
* <ul>
- * <li> Maintains two lists, one for md5s and another for schema. </li>
- * <li> MD5 at index i in the MD5 list, corresponds to Schema at index i in the schema list. </li>
+ * <li> Maintains two lists, one for hashes and another for schema. </li>
+ * <li> Hash at index i in the hash list corresponds to schema at index i in the schema list. </li>
* </ul>
*/
- private static class MD5AvroSchemaMap {
- private List<byte[]> _md5s;
- private List<org.apache.avro.Schema> _schemas;
+ private static class HashToSchemaMap {
+ private final List<byte[]> _hashes;
+ private final List<org.apache.avro.Schema> _schemas;
/**
* Constructor for the class.
*/
- private MD5AvroSchemaMap() {
- _md5s = new ArrayList<>();
+ private HashToSchemaMap() {
+ _hashes = new ArrayList<>();
_schemas = new ArrayList<>();
}
/**
- * Returns the Avro schema corresponding to the given MD5.
+ * Returns the Avro schema corresponding to the given hash.
*
- * @param md5ForSchema MD5 for which to get the avro schema.
- * @return Avro schema for the given MD5.
+ * @param hash Hash for which to get the avro schema.
+ * @return Avro schema for the given hash.
*/
- private org.apache.avro.Schema getSchema(byte[] md5ForSchema) {
- for (int i = 0; i < _md5s.size(); i++) {
- if (Arrays.equals(_md5s.get(i), md5ForSchema)) {
+ private org.apache.avro.Schema getSchema(byte[] hash) {
+ for (int i = 0; i < _hashes.size(); i++) {
+ if (Arrays.equals(_hashes.get(i), hash)) {
return _schemas.get(i);
}
}
@@ -282,14 +282,14 @@ public class KafkaAvroMessageDecoder implements StreamMessageDecoder<byte[]> {
}
/**
- * Adds mapping between MD5 and Avro schema.
- * Caller to ensure that addSchema is called only once per MD5-Schema pair.
+ * Adds mapping between hash and avro schema.
+ * Caller to ensure that addSchema is called only once per hash-schema pair.
*
- * @param md5 MD5 for the Schema
+ * @param hash Hash for the schema
* @param schema Avro Schema
*/
- private void addSchema(byte[] md5, org.apache.avro.Schema schema) {
- _md5s.add(Arrays.copyOf(md5, md5.length));
+ private void addSchema(byte[] hash, org.apache.avro.Schema schema) {
+ _hashes.add(Arrays.copyOf(hash, hash.length));
_schemas.add(schema);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org