Skip to content

Commit 74d7552

Browse files
bachuvbachuv
committed
initial commit
1 parent 6f996d1 commit 74d7552

40 files changed

Lines changed: 2491 additions & 0 deletions

‎client/src/main/java/com/microsoft/durabletask/DurableTaskClient.java‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,11 @@
22
// Licensed under the MIT License.
33
package com.microsoft.durabletask;
44

5+
import com.microsoft.durabletask.history.HistoryEvent;
6+
57
import javax.annotation.Nullable;
68
import java.time.Duration;
9+
import java.util.List;
710
import java.util.concurrent.TimeoutException;
811

912
/**
@@ -226,6 +229,31 @@ public abstract OrchestrationMetadata waitForInstanceCompletion(
226229
*/
227230
public abstract OrchestrationStatusQueryResult queryInstances(OrchestrationStatusQuery query);
228231

232+
/**
233+
* Lists the IDs of terminal orchestration instances that completed within a time window.
234+
* <p>
235+
* Unlike {@link #queryInstances(OrchestrationStatusQuery)}, which filters by creation time and returns full
236+
* metadata, this method filters by <em>completion</em> time and returns only instance IDs, making it efficient
237+
* for bulk enumeration such as archival/export. Results are paged; pass
238+
* {@link ListInstanceIdsResult#getContinuationToken()} back via
239+
* {@link ListInstanceIdsQuery#setContinuationToken(String)} to fetch subsequent pages.
240+
*
241+
* @param query filter criteria: completion-time window, terminal runtime statuses, page size, and pagination cursor
242+
* @return a page of matching instance IDs and a cursor for the next page
243+
*/
244+
public abstract ListInstanceIdsResult listInstanceIds(ListInstanceIdsQuery query);
245+
246+
/**
247+
* Gets the full history of an orchestration instance as an ordered list of {@link HistoryEvent} objects.
248+
* <p>
249+
* The events are returned in execution order. This is useful for archiving or offline analysis of an instance's
250+
* execution history. Use {@code instanceof} to inspect each concrete event type.
251+
*
252+
* @param instanceId the unique ID of the orchestration instance whose history to fetch
253+
* @return the instance's history events in order; empty if the instance has no history
254+
*/
255+
public abstract List<HistoryEvent> getOrchestrationHistory(String instanceId);
256+
229257
/**
230258
* Initializes the target task hub data store.
231259
* <p>

‎client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44

55
import com.google.protobuf.StringValue;
66
import com.google.protobuf.Timestamp;
7+
import com.microsoft.durabletask.history.HistoryEvent;
8+
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService;
79
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*;
810
import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
911
import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc.*;
@@ -295,6 +297,35 @@ private OrchestrationStatusQueryResult toQueryResult(QueryInstancesResponse quer
295297
return new OrchestrationStatusQueryResult(metadataList, queryInstancesResponse.getContinuationToken().getValue());
296298
}
297299

300+
@Override
301+
public ListInstanceIdsResult listInstanceIds(ListInstanceIdsQuery query) {
302+
ListInstanceIdsRequest.Builder builder = ListInstanceIdsRequest.newBuilder();
303+
Optional.ofNullable(query.getCompletedTimeFrom()).ifPresent(from -> builder.setCompletedTimeFrom(DataConverter.getTimestampFromInstant(from)));
304+
Optional.ofNullable(query.getCompletedTimeTo()).ifPresent(to -> builder.setCompletedTimeTo(DataConverter.getTimestampFromInstant(to)));
305+
String requestContinuationToken = query.getContinuationToken() != null ? query.getContinuationToken() : "";
306+
builder.setLastInstanceKey(StringValue.of(requestContinuationToken));
307+
builder.setPageSize(query.getPageSize());
308+
query.getRuntimeStatusList().forEach(runtimeStatus -> Optional.ofNullable(runtimeStatus).ifPresent(status -> builder.addRuntimeStatus(OrchestrationRuntimeStatus.toProtobuf(status))));
309+
ListInstanceIdsResponse response = this.sidecarClient.listInstanceIds(builder.build());
310+
String continuationToken = response.hasLastInstanceKey() ? response.getLastInstanceKey().getValue() : null;
311+
return new ListInstanceIdsResult(new ArrayList<>(response.getInstanceIdsList()), continuationToken);
312+
}
313+
314+
@Override
315+
public List<HistoryEvent> getOrchestrationHistory(String instanceId) {
316+
StreamInstanceHistoryRequest request = StreamInstanceHistoryRequest.newBuilder()
317+
.setInstanceId(instanceId)
318+
.build();
319+
List<HistoryEvent> historyEvents = new ArrayList<>();
320+
Iterator<HistoryChunk> chunks = this.sidecarClient.streamInstanceHistory(request);
321+
while (chunks.hasNext()) {
322+
for (OrchestratorService.HistoryEvent protoEvent : chunks.next().getEventsList()) {
323+
historyEvents.add(HistoryEventConverter.fromProto(protoEvent));
324+
}
325+
}
326+
return historyEvents;
327+
}
328+
298329
@Override
299330
public void createTaskHub(boolean recreateIfExists) {
300331
this.sidecarClient.createTaskHub(CreateTaskHubRequest.newBuilder().setRecreateIfExists(recreateIfExists).build());

‎client/src/main/java/com/microsoft/durabletask/HistoryEventConverter.java‎

Lines changed: 246 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,246 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
package com.microsoft.durabletask;
4+
5+
import com.microsoft.durabletask.history.*;
6+
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService;
7+
8+
import javax.annotation.Nullable;
9+
import java.time.Instant;
10+
11+
/**
12+
* Converts protobuf {@code HistoryEvent} messages into the public {@link HistoryEvent} domain model.
13+
*/
14+
final class HistoryEventConverter {
15+
private HistoryEventConverter() {
16+
}
17+
18+
/**
19+
* Converts a protobuf history event into its domain representation.
20+
*
21+
* @param proto the protobuf history event
22+
* @return the domain {@link HistoryEvent}
23+
* @throws IllegalArgumentException if the event type is not set
24+
* @throws UnsupportedOperationException if the event type is not recognized
25+
*/
26+
static HistoryEvent fromProto(OrchestratorService.HistoryEvent proto) {
27+
int id = proto.getEventId();
28+
Instant ts = DataConverter.getInstantFromTimestamp(proto.getTimestamp());
29+
switch (proto.getEventTypeCase()) {
30+
case EXECUTIONSTARTED: {
31+
OrchestratorService.ExecutionStartedEvent p = proto.getExecutionStarted();
32+
return new ExecutionStartedEvent(id, ts,
33+
p.getName(),
34+
stringOrNull(p.hasVersion(), p.getVersion()),
35+
stringOrNull(p.hasInput(), p.getInput()),
36+
p.hasOrchestrationInstance() ? toInstance(p.getOrchestrationInstance()) : null,
37+
p.hasParentInstance() ? toParentInfo(p.getParentInstance()) : null,
38+
p.hasScheduledStartTimestamp()
39+
? DataConverter.getInstantFromTimestamp(p.getScheduledStartTimestamp()) : null,
40+
p.hasParentTraceContext() ? toTrace(p.getParentTraceContext()) : null,
41+
stringOrNull(p.hasOrchestrationSpanID(), p.getOrchestrationSpanID()),
42+
p.getTagsMap());
43+
}
44+
case EXECUTIONCOMPLETED: {
45+
OrchestratorService.ExecutionCompletedEvent p = proto.getExecutionCompleted();
46+
return new ExecutionCompletedEvent(id, ts,
47+
OrchestrationRuntimeStatus.fromProtobuf(p.getOrchestrationStatus()),
48+
stringOrNull(p.hasResult(), p.getResult()),
49+
p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null);
50+
}
51+
case EXECUTIONTERMINATED: {
52+
OrchestratorService.ExecutionTerminatedEvent p = proto.getExecutionTerminated();
53+
return new ExecutionTerminatedEvent(id, ts, stringOrNull(p.hasInput(), p.getInput()), p.getRecurse());
54+
}
55+
case TASKSCHEDULED: {
56+
OrchestratorService.TaskScheduledEvent p = proto.getTaskScheduled();
57+
return new TaskScheduledEvent(id, ts,
58+
p.getName(),
59+
stringOrNull(p.hasVersion(), p.getVersion()),
60+
stringOrNull(p.hasInput(), p.getInput()),
61+
p.hasParentTraceContext() ? toTrace(p.getParentTraceContext()) : null,
62+
p.getTagsMap());
63+
}
64+
case TASKCOMPLETED: {
65+
OrchestratorService.TaskCompletedEvent p = proto.getTaskCompleted();
66+
return new TaskCompletedEvent(id, ts, p.getTaskScheduledId(), stringOrNull(p.hasResult(), p.getResult()));
67+
}
68+
case TASKFAILED: {
69+
OrchestratorService.TaskFailedEvent p = proto.getTaskFailed();
70+
return new TaskFailedEvent(id, ts, p.getTaskScheduledId(),
71+
p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null);
72+
}
73+
case SUBORCHESTRATIONINSTANCECREATED: {
74+
OrchestratorService.SubOrchestrationInstanceCreatedEvent p = proto.getSubOrchestrationInstanceCreated();
75+
return new SubOrchestrationInstanceCreatedEvent(id, ts,
76+
p.getInstanceId(),
77+
p.getName(),
78+
stringOrNull(p.hasVersion(), p.getVersion()),
79+
stringOrNull(p.hasInput(), p.getInput()),
80+
p.hasParentTraceContext() ? toTrace(p.getParentTraceContext()) : null,
81+
p.getTagsMap());
82+
}
83+
case SUBORCHESTRATIONINSTANCECOMPLETED: {
84+
OrchestratorService.SubOrchestrationInstanceCompletedEvent p =
85+
proto.getSubOrchestrationInstanceCompleted();
86+
return new SubOrchestrationInstanceCompletedEvent(id, ts,
87+
p.getTaskScheduledId(), stringOrNull(p.hasResult(), p.getResult()));
88+
}
89+
case SUBORCHESTRATIONINSTANCEFAILED: {
90+
OrchestratorService.SubOrchestrationInstanceFailedEvent p = proto.getSubOrchestrationInstanceFailed();
91+
return new SubOrchestrationInstanceFailedEvent(id, ts, p.getTaskScheduledId(),
92+
p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null);
93+
}
94+
case TIMERCREATED: {
95+
OrchestratorService.TimerCreatedEvent p = proto.getTimerCreated();
96+
return new TimerCreatedEvent(id, ts, DataConverter.getInstantFromTimestamp(p.getFireAt()));
97+
}
98+
case TIMERFIRED: {
99+
OrchestratorService.TimerFiredEvent p = proto.getTimerFired();
100+
return new TimerFiredEvent(id, ts, DataConverter.getInstantFromTimestamp(p.getFireAt()), p.getTimerId());
101+
}
102+
case ORCHESTRATORSTARTED:
103+
return new OrchestratorStartedEvent(id, ts);
104+
case ORCHESTRATORCOMPLETED:
105+
return new OrchestratorCompletedEvent(id, ts);
106+
case EVENTSENT: {
107+
OrchestratorService.EventSentEvent p = proto.getEventSent();
108+
return new EventSentEvent(id, ts, p.getInstanceId(), p.getName(),
109+
stringOrNull(p.hasInput(), p.getInput()));
110+
}
111+
case EVENTRAISED: {
112+
OrchestratorService.EventRaisedEvent p = proto.getEventRaised();
113+
return new EventRaisedEvent(id, ts, p.getName(), stringOrNull(p.hasInput(), p.getInput()));
114+
}
115+
case GENERICEVENT: {
116+
OrchestratorService.GenericEvent p = proto.getGenericEvent();
117+
return new GenericEvent(id, ts, stringOrNull(p.hasData(), p.getData()));
118+
}
119+
case HISTORYSTATE: {
120+
OrchestratorService.HistoryStateEvent p = proto.getHistoryState();
121+
return new HistoryStateEvent(id, ts,
122+
p.hasOrchestrationState() ? toOrchestrationState(p.getOrchestrationState()) : null);
123+
}
124+
case CONTINUEASNEW: {
125+
OrchestratorService.ContinueAsNewEvent p = proto.getContinueAsNew();
126+
return new ContinueAsNewEvent(id, ts, stringOrNull(p.hasInput(), p.getInput()));
127+
}
128+
case EXECUTIONSUSPENDED: {
129+
OrchestratorService.ExecutionSuspendedEvent p = proto.getExecutionSuspended();
130+
return new ExecutionSuspendedEvent(id, ts, stringOrNull(p.hasInput(), p.getInput()));
131+
}
132+
case EXECUTIONRESUMED: {
133+
OrchestratorService.ExecutionResumedEvent p = proto.getExecutionResumed();
134+
return new ExecutionResumedEvent(id, ts, stringOrNull(p.hasInput(), p.getInput()));
135+
}
136+
case ENTITYOPERATIONSIGNALED: {
137+
OrchestratorService.EntityOperationSignaledEvent p = proto.getEntityOperationSignaled();
138+
return new EntityOperationSignaledEvent(id, ts,
139+
p.getRequestId(),
140+
p.getOperation(),
141+
p.hasScheduledTime() ? DataConverter.getInstantFromTimestamp(p.getScheduledTime()) : null,
142+
stringOrNull(p.hasInput(), p.getInput()),
143+
stringOrNull(p.hasTargetInstanceId(), p.getTargetInstanceId()));
144+
}
145+
case ENTITYOPERATIONCALLED: {
146+
OrchestratorService.EntityOperationCalledEvent p = proto.getEntityOperationCalled();
147+
return new EntityOperationCalledEvent(id, ts,
148+
p.getRequestId(),
149+
p.getOperation(),
150+
p.hasScheduledTime() ? DataConverter.getInstantFromTimestamp(p.getScheduledTime()) : null,
151+
stringOrNull(p.hasInput(), p.getInput()),
152+
stringOrNull(p.hasParentInstanceId(), p.getParentInstanceId()),
153+
stringOrNull(p.hasParentExecutionId(), p.getParentExecutionId()),
154+
stringOrNull(p.hasTargetInstanceId(), p.getTargetInstanceId()));
155+
}
156+
case ENTITYOPERATIONCOMPLETED: {
157+
OrchestratorService.EntityOperationCompletedEvent p = proto.getEntityOperationCompleted();
158+
return new EntityOperationCompletedEvent(id, ts, p.getRequestId(),
159+
stringOrNull(p.hasOutput(), p.getOutput()));
160+
}
161+
case ENTITYOPERATIONFAILED: {
162+
OrchestratorService.EntityOperationFailedEvent p = proto.getEntityOperationFailed();
163+
return new EntityOperationFailedEvent(id, ts, p.getRequestId(),
164+
p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null);
165+
}
166+
case ENTITYLOCKREQUESTED: {
167+
OrchestratorService.EntityLockRequestedEvent p = proto.getEntityLockRequested();
168+
return new EntityLockRequestedEvent(id, ts,
169+
p.getCriticalSectionId(),
170+
p.getLockSetList(),
171+
p.getPosition(),
172+
stringOrNull(p.hasParentInstanceId(), p.getParentInstanceId()));
173+
}
174+
case ENTITYLOCKGRANTED: {
175+
OrchestratorService.EntityLockGrantedEvent p = proto.getEntityLockGranted();
176+
return new EntityLockGrantedEvent(id, ts, p.getCriticalSectionId());
177+
}
178+
case ENTITYUNLOCKSENT: {
179+
OrchestratorService.EntityUnlockSentEvent p = proto.getEntityUnlockSent();
180+
return new EntityUnlockSentEvent(id, ts,
181+
p.getCriticalSectionId(),
182+
stringOrNull(p.hasParentInstanceId(), p.getParentInstanceId()),
183+
stringOrNull(p.hasTargetInstanceId(), p.getTargetInstanceId()));
184+
}
185+
case EXECUTIONREWOUND: {
186+
OrchestratorService.ExecutionRewoundEvent p = proto.getExecutionRewound();
187+
return new ExecutionRewoundEvent(id, ts,
188+
stringOrNull(p.hasReason(), p.getReason()),
189+
stringOrNull(p.hasParentExecutionId(), p.getParentExecutionId()),
190+
stringOrNull(p.hasInstanceId(), p.getInstanceId()),
191+
p.hasParentTraceContext() ? toTrace(p.getParentTraceContext()) : null,
192+
stringOrNull(p.hasName(), p.getName()),
193+
stringOrNull(p.hasVersion(), p.getVersion()),
194+
stringOrNull(p.hasInput(), p.getInput()),
195+
p.hasParentInstance() ? toParentInfo(p.getParentInstance()) : null,
196+
p.getTagsMap());
197+
}
198+
case EVENTTYPE_NOT_SET:
199+
throw new IllegalArgumentException("History event does not have an eventType set.");
200+
default:
201+
throw new UnsupportedOperationException(
202+
"Deserialization of history event type " + proto.getEventTypeCase() + " is not supported.");
203+
}
204+
}
205+
206+
@Nullable
207+
private static String stringOrNull(boolean present, com.google.protobuf.StringValue value) {
208+
return present ? value.getValue() : null;
209+
}
210+
211+
private static OrchestrationInstance toInstance(OrchestratorService.OrchestrationInstance p) {
212+
return new OrchestrationInstance(p.getInstanceId(), stringOrNull(p.hasExecutionId(), p.getExecutionId()));
213+
}
214+
215+
private static ParentInstanceInfo toParentInfo(OrchestratorService.ParentInstanceInfo p) {
216+
return new ParentInstanceInfo(
217+
p.getTaskScheduledId(),
218+
stringOrNull(p.hasName(), p.getName()),
219+
stringOrNull(p.hasVersion(), p.getVersion()),
220+
p.hasOrchestrationInstance() ? toInstance(p.getOrchestrationInstance()) : null);
221+
}
222+
223+
private static TraceContext toTrace(OrchestratorService.TraceContext p) {
224+
return new TraceContext(p.getTraceParent(), stringOrNull(p.hasTraceState(), p.getTraceState()));
225+
}
226+
227+
private static OrchestrationState toOrchestrationState(OrchestratorService.OrchestrationState p) {
228+
return new OrchestrationState(
229+
p.getInstanceId(),
230+
p.getName(),
231+
stringOrNull(p.hasVersion(), p.getVersion()),
232+
OrchestrationRuntimeStatus.fromProtobuf(p.getOrchestrationStatus()),
233+
p.hasScheduledStartTimestamp()
234+
? DataConverter.getInstantFromTimestamp(p.getScheduledStartTimestamp()) : null,
235+
p.hasCreatedTimestamp() ? DataConverter.getInstantFromTimestamp(p.getCreatedTimestamp()) : null,
236+
p.hasLastUpdatedTimestamp() ? DataConverter.getInstantFromTimestamp(p.getLastUpdatedTimestamp()) : null,
237+
p.hasCompletedTimestamp() ? DataConverter.getInstantFromTimestamp(p.getCompletedTimestamp()) : null,
238+
stringOrNull(p.hasInput(), p.getInput()),
239+
stringOrNull(p.hasOutput(), p.getOutput()),
240+
stringOrNull(p.hasCustomStatus(), p.getCustomStatus()),
241+
p.hasFailureDetails() ? new FailureDetails(p.getFailureDetails()) : null,
242+
stringOrNull(p.hasExecutionId(), p.getExecutionId()),
243+
stringOrNull(p.hasParentInstanceId(), p.getParentInstanceId()),
244+
p.getTagsMap());
245+
}
246+
}

0 commit comments

Comments
 (0)