[GOBBLIN-2155] Fix bug where empty delete activity result was not jackson deserializ… (apache#4054)

* Fix bug where empty delete activity result was not jackson deserializable

Author: Will-Lo

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/activity/impl/DeleteWorkDirsActivityImpl.java

@@ -50,7 +50,7 @@ public DirDeletionResult delete(WUProcessingSpec workSpec, EventSubmitterContext
     // Ensure that non HDFS writers exit early as they rely on a different cleanup process, can consider consolidation in the future
     // through an abstracted cleanup method implemented at a writer level
     if (workDirPaths.isEmpty()) {
-      return new DirDeletionResult();
+      return DirDeletionResult.createEmpty();
     }
     //TODO: Emit timers to measure length of cleanup step
     Optional<String> optJobName = Optional.empty();

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/work/DirDeletionResult.java

@@ -17,6 +17,7 @@
 
 package org.apache.gobblin.temporal.ddm.work;
 
+import java.util.HashMap;
 import java.util.Map;
 
 import lombok.Data;
@@ -34,4 +35,16 @@
 public class DirDeletionResult {
 
   @NonNull private Map<String, Boolean> successesByDirPath;
+
+  /**
+   * Empty result that should be used instead of empty constructor and needed to support jackson (de)serialization, otherwise will face the following error
+   * Caused by: io.temporal.common.converter.DataConverterException: com.fasterxml.jackson.databind.JsonMappingException: successesByDirPath is marked non-null but is null
+   *  at [Source: (byte[])"{"successesByDirPath":null}"; line: 1, column: 23] (through reference chain: org.apache.gobblin.temporal.ddm.work.DirDeletionResult["successesByDirPath"])
+   * 	at io.temporal.common.converter.JacksonJsonPayloadConverter.fromData(JacksonJsonPayloadConverter.java:101)
+   * 	at io.temporal.common.converter.DefaultDataConverter.fromPayload(DefaultDataConverter.java:145)
+   * @return
+   */
+  public static DirDeletionResult createEmpty() {
+    return new DirDeletionResult(new HashMap<>());
+  }
 }

gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/activity/impl/DeleteWorkDirsActivityImplTest.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.temporal.ddm.activity.impl;
+
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import io.temporal.api.common.v1.Payload;
+import io.temporal.common.converter.JacksonJsonPayloadConverter;
+import io.temporal.common.converter.PayloadConverter;
+
+import org.apache.gobblin.temporal.ddm.activity.DeleteWorkDirsActivity;
+import org.apache.gobblin.temporal.ddm.work.DirDeletionResult;
+import org.apache.gobblin.temporal.ddm.work.WUProcessingSpec;
+
+
+public class DeleteWorkDirsActivityImplTest {
+
+  @Test
+  public void testEmptyDeleteSupportsSerde() {
+    DeleteWorkDirsActivity deleteWorkDirsActivity = new DeleteWorkDirsActivityImpl();
+    WUProcessingSpec workSpec = new WUProcessingSpec();
+    Set<String> workDirPaths = new HashSet<>();
+    DirDeletionResult result = deleteWorkDirsActivity.delete(workSpec, null, workDirPaths);
+    PayloadConverter converter = new JacksonJsonPayloadConverter();
+    Optional<Payload> payload = converter.toData(result);
+    DirDeletionResult result2 = converter.fromData(payload.get(), DirDeletionResult.class, DirDeletionResult.class);
+    Assert.assertEquals(result, result2);
+  }
+}