debugging, will clean up later.

Netflix · Jun 19, 2019 · de6c218 · de6c218
1 parent 9f71b13
commit de6c218
Show file tree

Hide file tree

Showing 18 changed files with 2,638 additions and 2,480 deletions.
diff --git a/core/src/main/java/com/netflix/conductor/core/config/Configuration.java b/core/src/main/java/com/netflix/conductor/core/config/Configuration.java
@@ -56,7 +56,7 @@ public interface Configuration {
     String ADDITIONAL_MODULES_PROPERTY_NAME = "conductor.additional.modules";
 
     String EXECUTION_LOCK_ENABLED_PROPERTY_NAME = "decider.locking.enabled";
-    boolean EXECUTION_LOCK_ENABLED_DEFAULT_VALUE = false;
+    boolean EXECUTION_LOCK_ENABLED_DEFAULT_VALUE = true;
 
     //TODO add constants for input/output external payload related properties.
 

diff --git a/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java b/core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java
@@ -877,7 +877,7 @@ public boolean decide(String workflowId) {
             LOGGER.error("Error deciding workflow: {}", workflowId, e);
             throw e;
         } finally {
-            executionLockService.releaseLock(workflowId);
+            executionLockService.releaseLock(workflowId, workflow.getStatus());
         }
         return false;
     }

diff --git a/core/src/main/java/com/netflix/conductor/core/utils/Lock.java b/core/src/main/java/com/netflix/conductor/core/utils/Lock.java
@@ -34,4 +34,5 @@ public interface Lock {
 
     void releaseLock(String lockId);
 
+    void deleteLock(String lockId);
 }
diff --git a/core/src/main/java/com/netflix/conductor/core/utils/NoopLock.java b/core/src/main/java/com/netflix/conductor/core/utils/NoopLock.java
@@ -14,4 +14,6 @@ public boolean acquireLock(String lockId, long timeToTry, TimeUnit unit) {
     @Override
     public void releaseLock(String lockId) {}
 
+    @Override
+    public void deleteLock(String lockId) {}
 }
diff --git a/core/src/main/java/com/netflix/conductor/service/ExecutionLockService.java b/core/src/main/java/com/netflix/conductor/service/ExecutionLockService.java
@@ -1,5 +1,6 @@
 package com.netflix.conductor.service;
 
+import com.netflix.conductor.common.run.Workflow;
 import com.netflix.conductor.core.config.Configuration;
 import com.netflix.conductor.core.utils.Lock;
 import com.netflix.conductor.metrics.Monitors;
@@ -25,7 +26,7 @@ public ExecutionLockService(Configuration config, Provider<Lock> lockProvider) {
 
     public boolean acquireLock(String lockId) {
         if (config.enableWorkflowExecutionLock()) {
-            Lock  lock = (Lock)lockProvider.get();
+            Lock lock = lockProvider.get();
             if (!lock.acquireLock(lockId, 2, TimeUnit.MILLISECONDS)) {
                 LOGGER.info("Thread {} failed to acquire lock to lockId {}.", Thread.currentThread().getId(), lockId);
                 Monitors.recordAcquireLockUnsuccessful(lockId);
@@ -42,17 +43,28 @@ public boolean acquireLock(String lockId) {
      */
     public void waitForLock(String lockId) {
         if (config.enableWorkflowExecutionLock()) {
-            Lock  lock = (Lock)lockProvider.get();
+            Lock  lock = lockProvider.get();
             lock.acquireLock(lockId);
             LOGGER.debug("Thread {} acquired lock to lockId {}.", Thread.currentThread().getId(), lockId);
         }
     }
 
     public void releaseLock(String lockId) {
         if (config.enableWorkflowExecutionLock()) {
-            Lock  lock = (Lock)lockProvider.get();
+            Lock  lock = lockProvider.get();
             lock.releaseLock(lockId);
             LOGGER.debug("Thread {} released lock to lockId {}.", Thread.currentThread().getId(), lockId);
         }
     }
+
+    public void releaseLock(String lockId, Workflow.WorkflowStatus status) {
+        if (config.enableWorkflowExecutionLock()) {
+            Lock  lock = lockProvider.get();
+            lock.releaseLock(lockId);
+//            if (status.isTerminal()) {
+//                lock.deleteLock(lockId);
+//            }
+//            LOGGER.debug("Thread {} deleted lockId {}.", Thread.currentThread().getId(), lockId);
+        }
+    }
 }
diff --git a/server/src/main/resources/log4j.properties b/server/src/main/resources/log4j.properties
@@ -15,11 +15,16 @@
 #
 
 # Set root logger level to DEBUG and its only appender to A1.
-log4j.rootLogger=DEBUG, A1
+log4j.rootLogger=INFO, A1
 
 # A1 is set to be a ConsoleAppender.
 log4j.appender.A1=org.apache.log4j.ConsoleAppender
 
+log4j.logger.org.apache.zookeeper=DEBUG
+log4j.logger.org.apache.curator=DEBUG
+log4j.logger.com.netflix.conductor.core.execution.WorkflowExecutor=DEBUG
+log4j.logger.com.netflix.conductor.service.ExecutionLockService=DEBUG
+
 # A1 uses PatternLayout.
 log4j.appender.A1.layout=org.apache.log4j.PatternLayout
 log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
diff --git a/server/src/main/resources/server.properties b/server/src/main/resources/server.properties
@@ -49,4 +49,4 @@ workflow.archive=false
 zk.connection=host1.2181,host2:2181,host3:2181
 
 #enable locking during workflow execution
-decider.locking.enabled=true
+decider.locking.enabled=false
diff --git a/...s/src/test/java/com/netflix/conductor/tests/integration/ESRestClientHttpEndToEndTest.java b/...s/src/test/java/com/netflix/conductor/tests/integration/ESRestClientHttpEndToEndTest.java
@@ -19,21 +19,22 @@
 import com.netflix.conductor.client.http.MetadataClient;
 import com.netflix.conductor.client.http.TaskClient;
 import com.netflix.conductor.client.http.WorkflowClient;
+import com.netflix.conductor.core.config.Configuration;
 import com.netflix.conductor.elasticsearch.ElasticSearchConfiguration;
 import com.netflix.conductor.elasticsearch.ElasticSearchRestClientProvider;
 import com.netflix.conductor.elasticsearch.EmbeddedElasticSearchProvider;
 import com.netflix.conductor.elasticsearch.SystemPropertiesElasticSearchConfiguration;
 import com.netflix.conductor.jetty.server.JettyServer;
 import com.netflix.conductor.tests.utils.TestEnvironment;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
 import org.elasticsearch.client.RestClient;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * @author Viren
  */
@@ -51,6 +52,7 @@ public static void setup() throws Exception {
         TestEnvironment.setup();
         System.setProperty(ElasticSearchConfiguration.EMBEDDED_PORT_PROPERTY_NAME, "9203");
         System.setProperty(ElasticSearchConfiguration.ELASTIC_SEARCH_URL_PROPERTY_NAME, "http://localhost:9203");
+        System.setProperty(Configuration.EXECUTION_LOCK_ENABLED_PROPERTY_NAME, "false");
 
         Injector bootInjector = Guice.createInjector(new BootstrapModule());
         Injector serverInjector = Guice.createInjector(bootInjector.getInstance(ModulesProvider.class).get());

diff --git a/...ness/src/test/java/com/netflix/conductor/tests/integration/ExclusiveJoinEndToEndTest.java b/...ness/src/test/java/com/netflix/conductor/tests/integration/ExclusiveJoinEndToEndTest.java
@@ -16,17 +16,6 @@
 
 package com.netflix.conductor.tests.integration;
 
-import static org.junit.Assert.assertEquals;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.netflix.conductor.bootstrap.BootstrapModule;
@@ -38,13 +27,24 @@
 import com.netflix.conductor.common.metadata.workflow.StartWorkflowRequest;
 import com.netflix.conductor.common.metadata.workflow.WorkflowDef;
 import com.netflix.conductor.common.run.Workflow;
+import com.netflix.conductor.core.config.Configuration;
 import com.netflix.conductor.elasticsearch.ElasticSearchConfiguration;
 import com.netflix.conductor.elasticsearch.EmbeddedElasticSearch;
 import com.netflix.conductor.elasticsearch.EmbeddedElasticSearchProvider;
 import com.netflix.conductor.jetty.server.JettyServer;
 import com.netflix.conductor.tests.integration.model.TaskWrapper;
 import com.netflix.conductor.tests.utils.JsonUtils;
 import com.netflix.conductor.tests.utils.TestEnvironment;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
 
 public class ExclusiveJoinEndToEndTest {
 
@@ -69,6 +69,7 @@ public static void setUp() throws Exception {
 		TestEnvironment.setup();
         System.setProperty(ElasticSearchConfiguration.EMBEDDED_PORT_PROPERTY_NAME, "9205");
         System.setProperty(ElasticSearchConfiguration.ELASTIC_SEARCH_URL_PROPERTY_NAME, "localhost:9305");
+		System.setProperty(Configuration.EXECUTION_LOCK_ENABLED_PROPERTY_NAME, "false");
 
 		Injector bootInjector = Guice.createInjector(new BootstrapModule());
 		Injector serverInjector = Guice.createInjector(bootInjector.getInstance(ModulesProvider.class).get());

diff --git a/test-harness/src/test/java/com/netflix/conductor/tests/integration/GrpcEndToEndTest.java b/test-harness/src/test/java/com/netflix/conductor/tests/integration/GrpcEndToEndTest.java
@@ -15,8 +15,6 @@
  */
 package com.netflix.conductor.tests.integration;
 
-import static org.junit.Assert.assertTrue;
-
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.netflix.conductor.bootstrap.BootstrapModule;
@@ -30,12 +28,14 @@
 import com.netflix.conductor.grpc.server.GRPCServer;
 import com.netflix.conductor.grpc.server.GRPCServerConfiguration;
 import com.netflix.conductor.grpc.server.GRPCServerProvider;
-import com.netflix.conductor.mysql.MySQLConfiguration;
 import com.netflix.conductor.tests.utils.TestEnvironment;
-import java.util.Optional;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
+import java.util.Optional;
+
+import static org.junit.Assert.assertTrue;
+
 /**
  * @author Viren
  */
@@ -50,6 +50,7 @@ public static void setup() throws Exception {
         System.setProperty(GRPCServerConfiguration.PORT_PROPERTY_NAME, "8092");
         System.setProperty(ElasticSearchConfiguration.EMBEDDED_PORT_PROPERTY_NAME, "9202");
         System.setProperty(ElasticSearchConfiguration.ELASTIC_SEARCH_URL_PROPERTY_NAME, "localhost:9302");
+        System.setProperty(Configuration.EXECUTION_LOCK_ENABLED_PROPERTY_NAME, "false");
 
         Injector bootInjector = Guice.createInjector(new BootstrapModule());
         Injector serverInjector = Guice.createInjector(bootInjector.getInstance(ModulesProvider.class).get());

diff --git a/test-harness/src/test/java/com/netflix/conductor/tests/integration/HttpEndToEndTest.java b/test-harness/src/test/java/com/netflix/conductor/tests/integration/HttpEndToEndTest.java
@@ -19,6 +19,7 @@
 import com.netflix.conductor.client.http.MetadataClient;
 import com.netflix.conductor.client.http.TaskClient;
 import com.netflix.conductor.client.http.WorkflowClient;
+import com.netflix.conductor.core.config.Configuration;
 import com.netflix.conductor.elasticsearch.ElasticSearchConfiguration;
 import com.netflix.conductor.elasticsearch.EmbeddedElasticSearchProvider;
 import com.netflix.conductor.jetty.server.JettyServer;
@@ -38,6 +39,7 @@ public static void setup() throws Exception {
         TestEnvironment.setup();
         System.setProperty(ElasticSearchConfiguration.EMBEDDED_PORT_PROPERTY_NAME, "9201");
         System.setProperty(ElasticSearchConfiguration.ELASTIC_SEARCH_URL_PROPERTY_NAME, "localhost:9301");
+        System.setProperty(Configuration.EXECUTION_LOCK_ENABLED_PROPERTY_NAME, "false");
 
         Injector bootInjector = Guice.createInjector(new BootstrapModule());
         Injector serverInjector = Guice.createInjector(bootInjector.getInstance(ModulesProvider.class).get());

diff --git a/...-harness/src/test/java/com/netflix/conductor/tests/integration/MySQLGrpcEndToEndTest.java b/...-harness/src/test/java/com/netflix/conductor/tests/integration/MySQLGrpcEndToEndTest.java
@@ -15,8 +15,6 @@
  */
 package com.netflix.conductor.tests.integration;
 
-import static org.junit.Assert.assertTrue;
-
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.netflix.conductor.bootstrap.BootstrapModule;
@@ -30,15 +28,16 @@
 import com.netflix.conductor.grpc.server.GRPCServer;
 import com.netflix.conductor.grpc.server.GRPCServerConfiguration;
 import com.netflix.conductor.grpc.server.GRPCServerProvider;
-import com.netflix.conductor.mysql.MySQLConfiguration;
-import com.netflix.conductor.tests.utils.MySQLTestModule;
 import com.netflix.conductor.tests.utils.MySQLTestRunner;
 import com.netflix.conductor.tests.utils.TestEnvironment;
-import java.util.Optional;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.runner.RunWith;
 
+import java.util.Optional;
+
+import static org.junit.Assert.assertTrue;
+
 /**
  * @author Viren
  */
@@ -54,6 +53,7 @@ public static void setup() throws Exception {
         System.setProperty(GRPCServerConfiguration.PORT_PROPERTY_NAME, "8094");
         System.setProperty(ElasticSearchConfiguration.EMBEDDED_PORT_PROPERTY_NAME, "9204");
         System.setProperty(ElasticSearchConfiguration.ELASTIC_SEARCH_URL_PROPERTY_NAME, "localhost:9304");
+        System.setProperty(Configuration.EXECUTION_LOCK_ENABLED_PROPERTY_NAME, "false");
 
         Injector bootInjector = Guice.createInjector(new BootstrapModule());
         Injector serverInjector = Guice.createInjector(bootInjector.getInstance(ModulesProvider.class).get());