ATLAS-1089: fix Storm hook to handle cyclic references in topology object

addons/storm-bridge/src/main/java/org/apache/atlas/storm/hook/StormAtlasHook.java

@@ -179,7 +179,7 @@ public class StormAtlasHook extends AtlasHook implements ISubmitterHook {
     private Referenceable createDataSet(String name, String topologyOwner,
                                               Serializable instance,
                                               Map stormConf, List<Referenceable> dependentEntities) throws IllegalAccessException {
-        Map<String, String> config = StormTopologyUtil.getFieldValues(instance, true);
+        Map<String, String> config = StormTopologyUtil.getFieldValues(instance, true, null);
 
         String clusterName = null;
         Referenceable dataSetReferenceable;
@@ -298,7 +298,7 @@ public class StormAtlasHook extends AtlasHook implements ISubmitterHook {
                 stormSpout.get_spout_object().get_serialized_java(), Serializable.class);
         spoutReferenceable.set("driverClass", instance.getClass().getName());
 
-        Map<String, String> flatConfigMap = StormTopologyUtil.getFieldValues(instance, true);
+        Map<String, String> flatConfigMap = StormTopologyUtil.getFieldValues(instance, true, null);
         spoutReferenceable.set("conf", flatConfigMap);
 
         return spoutReferenceable;
@@ -322,7 +322,7 @@ public class StormAtlasHook extends AtlasHook implements ISubmitterHook {
                 stormBolt.get_bolt_object().get_serialized_java(), Serializable.class);
         boltReferenceable.set("driverClass", instance.getClass().getName());
 
-        Map<String, String> flatConfigMap = StormTopologyUtil.getFieldValues(instance, true);
+        Map<String, String> flatConfigMap = StormTopologyUtil.getFieldValues(instance, true, null);
         boltReferenceable.set("conf", flatConfigMap);
 
         return boltReferenceable;

addons/storm-bridge/src/main/java/org/apache/atlas/storm/hook/StormTopologyUtil.java

@@ -127,10 +127,17 @@ public final class StormTopologyUtil {
     }
 
     public static Map<String, String> getFieldValues(Object instance,
-                                                     boolean prependClassName)
+                                                     boolean prependClassName,
+                                                     Set<Object> objectsToSkip)
     throws IllegalAccessException {
-        Class clazz = instance.getClass();
+        if (objectsToSkip == null) {
+            objectsToSkip = new HashSet<Object>();
+        }
+
         Map<String, String> output = new HashMap<>();
+
+        if (objectsToSkip.add(instance)) {
+            Class clazz = instance.getClass();
             for (Class<?> c = clazz; c != null; c = c.getSuperclass()) {
                 Field[] fields = c.getDeclaredFields();
                 for (Field field : fields) {
@@ -164,8 +171,8 @@ public final class StormTopologyUtil {
                             Object mapKey = ((Map.Entry) entry).getKey();
                             Object mapVal = ((Map.Entry) entry).getValue();
 
-                        String keyStr = getString(mapKey, false);
-                        String valStr = getString(mapVal, false);
+                            String keyStr = getString(mapKey, false, objectsToSkip);
+                            String valStr = getString(mapVal, false, objectsToSkip);
                             if ((valStr == null) || (valStr.isEmpty())) {
                                 continue;
                             } else {
@@ -176,17 +183,17 @@ public final class StormTopologyUtil {
                         //TODO check if it makes more sense to just stick to
                         // json like structure instead of a flatten output.
                         Collection collection = (Collection) fieldVal;
-                    if (collection.size()==0) continue;
+                        if (collection.size() == 0) continue;
                         String outStr = "";
                         for (Object o : collection) {
-                        outStr += getString(o, false) + ",";
+                            outStr += getString(o, false, objectsToSkip) + ",";
                         }
                         if (outStr.length() > 0) {
                             outStr = outStr.substring(0, outStr.length() - 1);
                         }
                         output.put(key, String.format("%s", outStr));
                     } else {
-                    Map<String, String> nestedFieldValues = getFieldValues(fieldVal, false);
+                        Map<String, String> nestedFieldValues = getFieldValues(fieldVal, false, objectsToSkip);
                         for (Map.Entry<String, String> entry : nestedFieldValues.entrySet()) {
                             output.put(String.format("%s.%s", key, entry.getKey()), entry.getValue());
                         }
@@ -196,17 +203,19 @@ public final class StormTopologyUtil {
                     }
                 }
             }
+        }
         return output;
     }
 
     private static String getString(Object instance,
-                                    boolean wrapWithQuote) throws IllegalAccessException {
+                                    boolean wrapWithQuote,
+                                    Set<Object> objectsToSkip) throws IllegalAccessException {
         if (instance == null) {
             return null;
         } else if (instance.getClass().isPrimitive() || isWrapperType(instance.getClass())) {
             return toString(instance, wrapWithQuote);
         } else {
-            return getString(getFieldValues(instance, false), wrapWithQuote);
+            return getString(getFieldValues(instance, false, objectsToSkip), wrapWithQuote);
         }
     }
 

release-log.txt

@@ -6,6 +6,7 @@ INCOMPATIBLE CHANGES:
 
 
 ALL CHANGES:
+ATLAS-1089 Storm hook should handle cyclic references in topology object (mneethiraj via sumasai)
 ATLAS-1086 Build failure in hive-bridge after security fixes in ATLAS-762 (sumasai)
 ATLAS-1088 Fix /search api to default to fulltext on dsl failure (sumasai)
 ATLAS-762 Assertion in NegativeSSLAndKerberosTest.testUnsecuredClient needs to be hardened (nixonrodrigues via sumasai)