add metric for corrupted signal

Author: halakaraki

src/main/java/com/uber/cadence/internal/metrics/MetricsType.java

@@ -121,6 +121,9 @@ public class MetricsType {
   public static final String UNHANDLED_SIGNALS_COUNTER =
       CADENCE_METRICS_PREFIX + "unhandled-signals";
 
+  public static final String CORRUPTED_SIGNALS_COUNTER =
+      CADENCE_METRICS_PREFIX + "corrupted-signals";
+
   public static final String WORKER_START_COUNTER = CADENCE_METRICS_PREFIX + "worker-start";
   public static final String POLLER_START_COUNTER = CADENCE_METRICS_PREFIX + "poller-start";
 

src/main/java/com/uber/cadence/internal/sync/POJOWorkflowImplementationFactory.java

@@ -23,6 +23,7 @@
 import com.uber.cadence.converter.DataConverterException;
 import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import com.uber.cadence.internal.common.InternalUtils;
+import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.replay.ReplayWorkflow;
 import com.uber.cadence.internal.replay.ReplayWorkflowFactory;
 import com.uber.cadence.internal.worker.WorkflowExecutionException;
@@ -35,6 +36,7 @@
 import com.uber.cadence.workflow.WorkflowInfo;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.cadence.workflow.WorkflowMethod;
+import com.uber.m3.tally.Scope;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
@@ -64,14 +66,17 @@ final class POJOWorkflowImplementationFactory implements ReplayWorkflowFactory {
       Collections.synchronizedMap(new HashMap<>());
 
   private final ExecutorService threadPool;
+  private final Scope metricsScope;
 
   POJOWorkflowImplementationFactory(
       DataConverter dataConverter,
       ExecutorService threadPool,
-      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory) {
+      Function<WorkflowInterceptor, WorkflowInterceptor> interceptorFactory,
+      Scope metricsScope) {
     this.dataConverter = Objects.requireNonNull(dataConverter);
     this.threadPool = Objects.requireNonNull(threadPool);
     this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
+    this.metricsScope = metricsScope;
   }
 
   void setWorkflowImplementationTypes(Class<?>[] workflowImplementationTypes) {
@@ -295,6 +300,7 @@ public void processSignal(String signalName, byte[] input, long eventId) {
                   + eventId
                   + ". Dropping it.",
               targetException);
+          metricsScope.counter(MetricsType.CORRUPTED_SIGNALS_COUNTER).inc(1);
         } else {
           throw new Error(
               "Failure processing \"" + signalName + "\" at eventID " + eventId, targetException);

src/main/java/com/uber/cadence/internal/sync/SyncWorkflowWorker.java

@@ -54,7 +54,10 @@ public SyncWorkflowWorker(
             new SynchronousQueue<>());
     factory =
         new POJOWorkflowImplementationFactory(
-            options.getDataConverter(), workflowThreadPool, interceptorFactory);
+            options.getDataConverter(),
+            workflowThreadPool,
+            interceptorFactory,
+            options.getMetricsScope());
     DecisionTaskHandler taskHandler = new ReplayDecisionTaskHandler(domain, factory, options);
     worker = new WorkflowWorker(service, domain, taskList, options, taskHandler);
     this.options = options;