ATLAS-1066 Falcon fails to post entity to Atlas due to kafka exception (mneethiraj via shwethags)

addons/falcon-bridge-shim/src/main/java/org/apache/atlas/falcon/service/AtlasService.java

@@ -21,6 +21,7 @@ package org.apache.atlas.falcon.service;
 
 import org.apache.atlas.plugin.classloader.AtlasPluginClassLoader;
 import org.apache.falcon.FalconException;
+import org.apache.falcon.entity.store.ConfigurationStore;
 import org.apache.falcon.entity.v0.Entity;
 import org.apache.falcon.service.ConfigurationChangeListener;
 import org.apache.falcon.service.FalconService;
@@ -74,6 +75,9 @@ public class AtlasService implements FalconService, ConfigurationChangeListener
 
         try {
             activatePluginClassLoader();
+
+            ConfigurationStore.get().registerListener(this);
+
             falconServiceImpl.init();
         } finally {
             deactivatePluginClassLoader();
@@ -92,6 +96,9 @@ public class AtlasService implements FalconService, ConfigurationChangeListener
 
         try {
             activatePluginClassLoader();
+
+            ConfigurationStore.get().unregisterListener(this);
+
             falconServiceImpl.destroy();
         } finally {
             deactivatePluginClassLoader();

addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/hook/FalconHook.java

@@ -136,7 +136,7 @@ public class FalconHook extends AtlasHook implements FalconEventPublisher {
                 });
             }
         } catch (Throwable t) {
-            LOG.warn("Error in processing data {}", data);
+            LOG.warn("Error in processing data " + data, t);
         }
     }
 

addons/falcon-bridge/src/main/java/org/apache/atlas/falcon/service/AtlasService.java

@@ -23,7 +23,6 @@ import org.apache.atlas.falcon.event.FalconEvent;
 import org.apache.atlas.falcon.hook.FalconHook;
 import org.apache.atlas.falcon.publisher.FalconEventPublisher;
 import org.apache.falcon.FalconException;
-import org.apache.falcon.entity.store.ConfigurationStore;
 import org.apache.falcon.entity.v0.Entity;
 import org.apache.falcon.entity.v0.EntityType;
 import org.apache.falcon.service.ConfigurationChangeListener;
@@ -51,13 +50,11 @@ public class AtlasService implements FalconService, ConfigurationChangeListener
 
     @Override
     public void init() throws FalconException {
-        ConfigurationStore.get().registerListener(this);
         publisher = new FalconHook();
     }
 
     @Override
     public void destroy() throws FalconException {
-        ConfigurationStore.get().unregisterListener(this);
     }
 
     @Override

release-log.txt

@@ -6,6 +6,7 @@ INCOMPATIBLE CHANGES:
 
 
 ALL CHANGES:
+ATLAS-1066 Falcon fails to post entity to Atlas due to kafka exception (mneethiraj via shwethags)
 ATLAS-1064 UI: Pagination for full text search results (Kalyanikashikar via shwethags)
 ATLAS-1006 Paginate full text search results (shwethags)
 ATLAS-1046 UI: Search pagination refinements (Kalyanikashikar via sumasai)