Merge pull request apache#32 from PraveenAdlakha/inmobi_0.10

FALCON-719,722,723 fixed
sandeepSamudrala · Oct 20, 2016 · ad5ad64 · ad5ad64
2 parents 0a204b1 + 09afe53
commit ad5ad64
Show file tree

Hide file tree

Showing 6 changed files with 24 additions and 10 deletions.
diff --git a/common/src/main/resources/startup.properties b/common/src/main/resources/startup.properties
@@ -346,5 +346,7 @@ it.workflow.execution.listeners=org.apache.falcon.catalog.CatalogPartitionHandle
 *.falcon.postprocessing.enable=true
 
 ### LogMoveService Properties
-*.falcon.logMoveService.threadCount=200
+*.falcon.logMoveService.max.threadCount=200
 *.falcon.logMoveService.blockingQueue.length=50
+##Note min threadCount should always be smaller than max threadCount.
+*.falcon.logMoveService.min.threadCount=20
diff --git a/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java b/messaging/src/main/java/org/apache/falcon/messaging/JMSMessageConsumer.java
@@ -156,7 +156,8 @@ private WorkflowExecutionContext createContext(TextMessage message) throws JMSEx
             wfProperties.put(WorkflowExecutionArgs.ENTITY_TYPE, entityTypePair.second.name());
             wfProperties.put(WorkflowExecutionArgs.WORKFLOW_USER, message.getStringProperty("user"));
             wfProperties.put(WorkflowExecutionArgs.OPERATION, getOperation(appName).name());
-
+            wfProperties.put(WorkflowExecutionArgs.USER_SUBFLOW_ID,
+                    json.getString("id").concat("@user-action"));
             String appType = message.getStringProperty("appType");
             return WorkflowExecutionContext.create(wfProperties, WorkflowExecutionContext.Type.valueOf(appType));
 

diff --git a/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java b/oozie/src/main/java/org/apache/falcon/logging/JobLogMover.java
@@ -92,7 +92,6 @@ public int run(WorkflowExecutionContext context) {
                 LOG.error("Error getting jobinfo for: {}", context.getUserSubflowId(), e);
                 return 0;
             }
-
             //Assumption is - Each wf run will have a directory
             //the corresponding job logs are stored within the respective dir
             Path path = new Path(context.getLogDir() + "/"

diff --git a/oozie/src/main/java/org/apache/falcon/service/LogMoverService.java b/oozie/src/main/java/org/apache/falcon/service/LogMoverService.java
@@ -46,14 +46,24 @@ public class LogMoverService implements WorkflowExecutionListener  {
 
     private BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(Integer.parseInt(
             StartupProperties.get().getProperty("falcon.logMoveService.blockingQueue.length", "50")));
-    private ExecutorService executorService = new ThreadPoolExecutor(20, getThreadCount(), 120,
+    private ExecutorService executorService = new ThreadPoolExecutor(getCorePoolSize(), getThreadCount(), 120,
             TimeUnit.SECONDS, blockingQueue);
+
+    public int getCorePoolSize(){
+        try{
+            return Integer.parseInt(StartupProperties.get().getProperty("falcon.logMoveService.min.threadCount", "20"));
+        } catch (NumberFormatException  e){
+            LOG.error("Exception in LogMoverService", e);
+            return 20;
+        }
+    }
     public int getThreadCount() {
         try{
-            return Integer.parseInt(StartupProperties.get().getProperty("falcon.logMoveService.threadCount", "200"));
+            return Integer.parseInt(StartupProperties.get()
+                    .getProperty("falcon.logMoveService.max.threadCount", "200"));
         } catch (NumberFormatException  e){
             LOG.error("Exception in LogMoverService", e);
-            return 50;
+            return 200;
         }
     }
 
@@ -86,7 +96,7 @@ private void onEnd(WorkflowExecutionContext context){
         if (Boolean.parseBoolean(ENABLE_POSTPROCESSING)) {
             return;
         }
-        while(0<blockingQueue.remainingCapacity()){
+        while(blockingQueue.remainingCapacity()<=0){
             try {
                 LOG.trace("Sleeping, no capacity in threadpool....");
                 TimeUnit.MILLISECONDS.sleep(500);

diff --git a/prism/src/main/java/org/apache/falcon/plugin/GraphiteNotificationPlugin.java b/prism/src/main/java/org/apache/falcon/plugin/GraphiteNotificationPlugin.java
@@ -50,9 +50,9 @@ public void monitor(ResourceMessage message) {
             String entityName = StringUtils.isNotBlank(message.getDimensions().get("entityName"))
                     ? message.getDimensions().get("entityName") :message.getDimensions().get("entity-name");
             String prefix = StartupProperties.get().getProperty("falcon.graphite.prefix");
-            String separator = ".";
             LOG.debug("message:" + message.getAction());
-            if (entityType.equalsIgnoreCase(EntityType.PROCESS.name())) {
+            if (entityType.equalsIgnoreCase(EntityType.PROCESS.name())
+                    && ConfigurationStore.get().get(EntityType.PROCESS, entityName) != null) {
                 Entity entity = ConfigurationStore.get().get(EntityType.PROCESS, entityName);
                 Process process = (Process) entity;
                 String pipeline =  StringUtils.isNotBlank(process.getPipelines()) ? process.getPipelines() : "default";

diff --git a/src/conf/startup.properties b/src/conf/startup.properties
@@ -363,5 +363,7 @@ prism.configstore.listeners=org.apache.falcon.entity.v0.EntityGraph,\
 *.falcon.postprocessing.enable=true
 
 ### LogMoveService Properties
-*.falcon.logMoveService.threadCount=200
+*.falcon.logMoveService.max.threadCount=200
 *.falcon.logMoveService.blockingQueue.length=50
+##Note min threadCount should always be smaller than max threadCount.
+*.falcon.logMoveService.min.threadCount=20