From fef2e2fc88c32476d42e79ab2b8d518c1afd9ca7 Mon Sep 17 00:00:00 2001
From: Ajay Kumar <ajay@apache.org>
Date: Wed, 27 Feb 2019 10:40:03 -0800
Subject: [PATCH] HDDS-1176. Allow persisting X509CertImpl to SCM certificate
 table. Contributed by Xiaoyu Yao.

---
 .../apache/hadoop/utils/db/CodecRegistry.java | 40 ++++++++++++++-----
 1 file changed, 31 insertions(+), 9 deletions(-)

diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/CodecRegistry.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/CodecRegistry.java
index afa72ad6390..1dd18420b10 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/CodecRegistry.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/utils/db/CodecRegistry.java
@@ -44,16 +44,13 @@ public class CodecRegistry {
    * @param <T>     Type of the return value.
    * @return the object with the parsed field data
    */
-  public <T> T asObject(byte[] rawData, Class<T> format) throws IOException {
+  public <T> T asObject(byte[] rawData, Class<T> format)
+      throws IOException {
     if (rawData == null) {
       return null;
     }
-    if (valueCodecs.containsKey(format)) {
-      return (T) valueCodecs.get(format).fromPersistedFormat(rawData);
-    } else {
-      throw new IllegalStateException(
-          "Codec is not registered for type: " + format);
-    }
+    Codec codec = getCodec(format);
+    return (T) codec.fromPersistedFormat(rawData);
   }
 
   /**
@@ -66,14 +63,39 @@ public class CodecRegistry {
   public <T> byte[] asRawData(T object) throws IOException {
     Preconditions.checkNotNull(object,
         "Null value shouldn't be persisted in the database");
+    Codec<T> codec = getCodec(object);
+    return codec.toPersistedFormat(object);
+  }
+
+  /**
+   * Get codec for the typed object including class and subclass.
+   * @param object typed object.
+   * @return Codec for the typed object.
+   * @throws IOException
+   */
+  private <T> Codec getCodec(T object) throws IOException {
     Class<T> format = (Class<T>) object.getClass();
+    return getCodec(format);
+  }
+
+
+  /**
+   * Get codec for the typed object including class and subclass.
+   * @param <T>    Type of the typed object.
+   * @return Codec for the typed object.
+   * @throws IOException
+   */
+  private <T> Codec getCodec(Class<T> format) throws IOException {
+    Codec<T> codec;
     if (valueCodecs.containsKey(format)) {
-      Codec<T> codec = (Codec<T>) valueCodecs.get(format);
-      return codec.toPersistedFormat(object);
+      codec = (Codec<T>) valueCodecs.get(format);
+    } else if (valueCodecs.containsKey(format.getSuperclass())) {
+      codec = (Codec<T>) valueCodecs.get(format.getSuperclass());
     } else {
       throw new IllegalStateException(
           "Codec is not registered for type: " + format);
     }
+    return codec;
   }
 
   /**