Revert "Revert "Integration get raw history #443 (#472)""

This reverts commit 7cbfb7b.

Author: Liang Mei

src/main/idls

@@ -18,19 +18,30 @@
 package com.uber.cadence.internal.common;
 
 import com.google.common.base.Defaults;
+import com.uber.cadence.DataBlob;
+import com.uber.cadence.History;
+import com.uber.cadence.HistoryEvent;
+import com.uber.cadence.HistoryEventFilterType;
 import com.uber.cadence.SearchAttributes;
 import com.uber.cadence.TaskList;
 import com.uber.cadence.TaskListKind;
+import com.uber.cadence.EntityNotExistsError;
 import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.converter.JsonDataConverter;
 import com.uber.cadence.internal.worker.Shutdownable;
 import com.uber.cadence.workflow.WorkflowMethod;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TException;
+import org.apache.thrift.TSerializer;
 
 /** Utility functions shared by the implementation code. */
 public final class InternalUtils {
@@ -143,6 +154,56 @@ public static SearchAttributes convertMapToSearchAttributes(
     return new SearchAttributes().setIndexedFields(mapOfByteBuffer);
   }
 
+  // This method deserialize the DataBlob data to the HistoriyEvent data
+  public static History DeserializeFromBlobToHistoryEvents(
+      List<DataBlob> blobData, HistoryEventFilterType historyEventFilterType) throws TException {
+
+    List<HistoryEvent> events = new ArrayList<HistoryEvent>();
+    for (DataBlob data : blobData) {
+      History history = new History();
+      try {
+        byte[] dataByte = data.getData();
+        dataByte = Arrays.copyOfRange(dataByte, 1, dataByte.length);
+        deSerializer.deserialize(history, dataByte);
+
+        if (history == null || history.getEvents() == null || history.getEvents().size() == 0) {
+          return null;
+        }
+      } catch (org.apache.thrift.TException err) {
+        throw new TException("Deserialize blob data to history event failed with unknown error");
+      }
+
+      events.addAll(history.getEvents());
+    }
+
+    if (events.size() > 0 && historyEventFilterType == HistoryEventFilterType.CLOSE_EVENT) {
+      events = events.subList(events.size() - 1, events.size());
+    }
+
+    return new History().setEvents(events);
+  }
+
+  // This method deserialize the history event data to blob data
+  public static List<DataBlob> DeserializeFromHistoryEventToBlobData(List<HistoryEvent> events)
+      throws EntityNotExistsError {
+
+    List<DataBlob> blobs = new ArrayList<DataBlob>();
+    for (HistoryEvent event : events) {
+      DataBlob blob = new DataBlob();
+      try {
+        blob.setData(serializer.serialize(event));
+      } catch (org.apache.thrift.TException err) {
+        throw new EntityNotExistsError("Deserialize blob data to history event failed with unknown error");
+      }
+      blobs.add(blob);
+    }
+
+    return blobs;
+  }
+
+  private static final TDeserializer deSerializer = new TDeserializer();
+  private static final TSerializer serializer = new TSerializer();
+
   /** Prohibit instantiation */
   private InternalUtils() {}
 }

src/main/java/com/uber/cadence/internal/common/InternalUtils.java

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import org.apache.thrift.TException;
 
 interface TestWorkflowStore {
 
@@ -155,8 +156,7 @@ void sendQueryTask(ExecutionId executionId, TaskListId taskList, PollForDecision
       throws EntityNotExistsError;
 
   GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
-      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest)
-      throws EntityNotExistsError;
+      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest) throws EntityNotExistsError;
 
   void getDiagnostics(StringBuilder result);
 

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStore.java

@@ -20,6 +20,8 @@
 import com.uber.cadence.BadRequestError;
 import com.uber.cadence.EntityNotExistsError;
 import com.uber.cadence.EventType;
+import com.uber.cadence.DataBlob;
+import com.uber.cadence.internal.common.InternalUtils;
 import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
 import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
 import com.uber.cadence.History;
@@ -47,6 +49,7 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import org.apache.thrift.TException;
 
 class TestWorkflowStoreImpl implements TestWorkflowStore {
 
@@ -335,8 +338,7 @@ public void sendQueryTask(
 
   @Override
   public GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
-      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest)
-      throws EntityNotExistsError {
+      ExecutionId executionId, GetWorkflowExecutionHistoryRequest getRequest) throws EntityNotExistsError {
     HistoryStore history;
     // Used to eliminate the race condition on waitForNewEvents
     long expectedNextEventId;
@@ -346,20 +348,23 @@ public GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
       if (!getRequest.isWaitForNewEvent()
           && getRequest.getHistoryEventFilterType() != HistoryEventFilterType.CLOSE_EVENT) {
         List<HistoryEvent> events = history.getEventsLocked();
+        List<DataBlob> blobs = InternalUtils.DeserializeFromHistoryEventToBlobData(events);
         // Copy the list as it is mutable. Individual events assumed immutable.
         ArrayList<HistoryEvent> eventsCopy = new ArrayList<>(events);
         return new GetWorkflowExecutionHistoryResponse()
-            .setHistory(new History().setEvents(eventsCopy));
+            .setHistory(new History().setEvents(eventsCopy)).setRawHistory(blobs);
       }
       expectedNextEventId = history.getNextEventIdLocked();
     } finally {
       lock.unlock();
     }
     List<HistoryEvent> events =
         history.waitForNewEvents(expectedNextEventId, getRequest.getHistoryEventFilterType());
+    List<DataBlob> blobs = InternalUtils.DeserializeFromHistoryEventToBlobData(events);
     GetWorkflowExecutionHistoryResponse result = new GetWorkflowExecutionHistoryResponse();
     if (events != null) {
       result.setHistory(new History().setEvents(events));
+      result.setRawHistory(blobs);
     }
     return result;
   }

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStoreImpl.java

@@ -37,6 +37,7 @@
 import com.uber.cadence.GetSearchAttributesResponse;
 import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
 import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
+import com.uber.cadence.History;
 import com.uber.cadence.InternalServiceError;
 import com.uber.cadence.LimitExceededError;
 import com.uber.cadence.ListArchivedWorkflowExecutionsRequest;
@@ -90,6 +91,7 @@
 import com.uber.cadence.WorkflowService.GetWorkflowExecutionHistory_result;
 import com.uber.cadence.internal.Version;
 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.metrics.NoopScope;
 import com.uber.cadence.internal.metrics.ServiceMethod;
@@ -106,10 +108,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
+import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import org.apache.thrift.TException;
@@ -853,7 +852,14 @@ private GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
       WorkflowService.GetWorkflowExecutionHistory_result result =
           response.getBody(WorkflowService.GetWorkflowExecutionHistory_result.class);
       if (response.getResponseCode() == ResponseCode.OK) {
-        return result.getSuccess();
+        GetWorkflowExecutionHistoryResponse res = result.getSuccess();
+        if (res.getRawHistory() != null) {
+          History history =
+              InternalUtils.DeserializeFromBlobToHistoryEvents(
+                  res.getRawHistory(), getRequest.getHistoryEventFilterType());
+          res.setHistory(history);
+        }
+        return res;
       }
       if (result.isSetBadRequestError()) {
         throw result.getBadRequestError();