Skip to content

Commit c6467bd

Browse files
bachuvbachuv
committed
addressed pr review comments
1 parent 7c55e0f commit c6467bd

11 files changed

Lines changed: 845 additions & 26 deletions

File tree

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

Lines changed: 34 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,10 @@
88
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.HistoryEvent;
99

1010
import java.time.Instant;
11+
import java.util.ArrayList;
1112
import java.util.HashMap;
1213
import java.util.LinkedHashMap;
14+
import java.util.List;
1315
import java.util.Map;
1416

1517
/**
@@ -99,8 +101,13 @@ private static Map<String, Object> extractEventData(
99101
Descriptors.FieldDescriptor field = entry.getKey();
100102
Object value = entry.getValue();
101103

102-
// Convert proto types to simple Java types
103-
data.put(field.getJsonName(), convertProtoValue(value));
104+
// Proto map fields appear as repeated MapEntry messages in reflection;
105+
// convert them to plain Java Maps using the field descriptor.
106+
if (field.isMapField()) {
107+
data.put(field.getJsonName(), convertProtoMapField(field, value));
108+
} else {
109+
data.put(field.getJsonName(), convertProtoValue(value));
110+
}
104111
}
105112

106113
return data;
@@ -140,6 +147,22 @@ private static Message getEventMessage(HistoryEvent proto, HistoryEvent.EventTyp
140147
}
141148
}
142149

150+
@SuppressWarnings("unchecked")
151+
private static Map<String, Object> convertProtoMapField(Descriptors.FieldDescriptor field, Object value) {
152+
// Proto map fields are represented as List<MapEntry> in reflection.
153+
// Each MapEntry is a Message with "key" and "value" fields.
154+
List<Message> entries = (List<Message>) value;
155+
Map<String, Object> result = new LinkedHashMap<>();
156+
Descriptors.FieldDescriptor keyField = field.getMessageType().findFieldByName("key");
157+
Descriptors.FieldDescriptor valueField = field.getMessageType().findFieldByName("value");
158+
for (Message entry : entries) {
159+
String key = String.valueOf(entry.getField(keyField));
160+
Object val = convertProtoValue(entry.getField(valueField));
161+
result.put(key, val);
162+
}
163+
return result;
164+
}
165+
143166
private static Object convertProtoValue(Object value) {
144167
if (value instanceof Timestamp) {
145168
Timestamp ts = (Timestamp) value;
@@ -154,6 +177,15 @@ private static Object convertProtoValue(Object value) {
154177
if (value instanceof com.google.protobuf.ProtocolMessageEnum) {
155178
return ((com.google.protobuf.ProtocolMessageEnum) value).getValueDescriptor().getName();
156179
}
180+
if (value instanceof List) {
181+
// Repeated fields come through as List<Object> from proto reflection
182+
List<?> protoList = (List<?>) value;
183+
List<Object> converted = new ArrayList<>(protoList.size());
184+
for (Object item : protoList) {
185+
converted.add(convertProtoValue(item));
186+
}
187+
return converted;
188+
}
157189
if (value instanceof Message) {
158190
// Recursively flatten nested messages
159191
Map<String, Object> nested = new LinkedHashMap<>();

‎client/src/test/java/com/microsoft/durabletask/OrchestrationHistoryEventMapperTest.java‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@
77
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*;
88
import org.junit.jupiter.api.Test;
99

10+
import java.util.List;
11+
import java.util.Map;
12+
1013
import static org.junit.jupiter.api.Assertions.*;
1114

1215
/**
@@ -152,4 +155,113 @@ void fromProto_timestampConversion_isCorrect() {
152155
assertEquals(1704067200, event.getTimestamp().getEpochSecond());
153156
assertEquals(500000000, event.getTimestamp().getNano());
154157
}
158+
159+
@SuppressWarnings("unchecked")
160+
@Test
161+
void fromProto_taskScheduledWithTags_mapsTagsAsMap() {
162+
HistoryEvent proto = HistoryEvent.newBuilder()
163+
.setEventId(10)
164+
.setTimestamp(Timestamp.newBuilder().setSeconds(1704067200).build())
165+
.setTaskScheduled(TaskScheduledEvent.newBuilder()
166+
.setName("MyActivity")
167+
.putTags("env", "production")
168+
.putTags("region", "westus2")
169+
.build())
170+
.build();
171+
172+
OrchestrationHistoryEvent event = OrchestrationHistoryEventMapper.fromProto(proto);
173+
174+
assertEquals("TaskScheduled", event.getEventType());
175+
Object tags = event.getData().get("tags");
176+
assertNotNull(tags, "tags should be present in data");
177+
assertInstanceOf(Map.class, tags, "tags should be a Map");
178+
Map<String, Object> tagsMap = (Map<String, Object>) tags;
179+
assertEquals("production", tagsMap.get("env"));
180+
assertEquals("westus2", tagsMap.get("region"));
181+
}
182+
183+
@SuppressWarnings("unchecked")
184+
@Test
185+
void fromProto_entityLockRequested_mapsLockSetAsList() {
186+
HistoryEvent proto = HistoryEvent.newBuilder()
187+
.setEventId(11)
188+
.setTimestamp(Timestamp.newBuilder().setSeconds(1704067200).build())
189+
.setEntityLockRequested(EntityLockRequestedEvent.newBuilder()
190+
.setCriticalSectionId("cs-1")
191+
.addLockSet("entity1")
192+
.addLockSet("entity2")
193+
.addLockSet("entity3")
194+
.build())
195+
.build();
196+
197+
OrchestrationHistoryEvent event = OrchestrationHistoryEventMapper.fromProto(proto);
198+
199+
assertEquals("EntityLockRequested", event.getEventType());
200+
Object lockSet = event.getData().get("lockSet");
201+
assertNotNull(lockSet, "lockSet should be present in data");
202+
assertInstanceOf(List.class, lockSet, "lockSet should be a List");
203+
List<Object> lockList = (List<Object>) lockSet;
204+
assertEquals(3, lockList.size());
205+
assertEquals("entity1", lockList.get(0));
206+
assertEquals("entity2", lockList.get(1));
207+
assertEquals("entity3", lockList.get(2));
208+
}
209+
210+
@SuppressWarnings("unchecked")
211+
@Test
212+
void fromProto_executionStartedWithTags_mapsTagsAsMap() {
213+
HistoryEvent proto = HistoryEvent.newBuilder()
214+
.setEventId(12)
215+
.setTimestamp(Timestamp.newBuilder().setSeconds(1704067200).build())
216+
.setExecutionStarted(ExecutionStartedEvent.newBuilder()
217+
.setName("MyOrch")
218+
.putTags("owner", "team-a")
219+
.build())
220+
.build();
221+
222+
OrchestrationHistoryEvent event = OrchestrationHistoryEventMapper.fromProto(proto);
223+
224+
assertEquals("ExecutionStarted", event.getEventType());
225+
Object tags = event.getData().get("tags");
226+
assertNotNull(tags, "tags should be present");
227+
assertInstanceOf(Map.class, tags);
228+
Map<String, Object> tagsMap = (Map<String, Object>) tags;
229+
assertEquals("team-a", tagsMap.get("owner"));
230+
}
231+
232+
@Test
233+
void fromProto_emptyRepeatedField_notIncludedInData() {
234+
// Proto repeated fields with no elements are not included in getAllFields()
235+
HistoryEvent proto = HistoryEvent.newBuilder()
236+
.setEventId(13)
237+
.setTimestamp(Timestamp.newBuilder().setSeconds(1704067200).build())
238+
.setEntityLockRequested(EntityLockRequestedEvent.newBuilder()
239+
.setCriticalSectionId("cs-empty")
240+
.build())
241+
.build();
242+
243+
OrchestrationHistoryEvent event = OrchestrationHistoryEventMapper.fromProto(proto);
244+
245+
// lockSet should not be present since it's empty (proto omits default-valued fields)
246+
assertFalse(event.getData().containsKey("lockSet"),
247+
"Empty repeated field should not appear in data");
248+
}
249+
250+
@SuppressWarnings("unchecked")
251+
@Test
252+
void fromProto_emptyMapField_notIncludedInData() {
253+
// Proto map fields with no entries are not included in getAllFields()
254+
HistoryEvent proto = HistoryEvent.newBuilder()
255+
.setEventId(14)
256+
.setTimestamp(Timestamp.newBuilder().setSeconds(1704067200).build())
257+
.setTaskScheduled(TaskScheduledEvent.newBuilder()
258+
.setName("NoTagsActivity")
259+
.build())
260+
.build();
261+
262+
OrchestrationHistoryEvent event = OrchestrationHistoryEventMapper.fromProto(proto);
263+
264+
assertFalse(event.getData().containsKey("tags"),
265+
"Empty map field should not appear in data");
266+
}
155267
}

‎exporthistory/build.gradle‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,14 @@
11
plugins {
22
id 'java-library'
33
id 'maven-publish'
4+
id 'signing'
5+
id 'com.github.spotbugs' version '6.4.8'
46
}
57

8+
group 'com.microsoft'
9+
version = '1.9.0'
10+
archivesBaseName = 'durabletask-exporthistory'
11+
612
ext {
713
azureStorageBlobVersion = '12.29.1'
814
jacksonVersion = '2.18.3'
@@ -28,3 +34,93 @@ dependencies {
2834
test {
2935
useJUnitPlatform()
3036
}
37+
38+
publishing {
39+
repositories {
40+
maven {
41+
url "file://$project.rootDir/repo"
42+
}
43+
}
44+
publications {
45+
mavenJava(MavenPublication) {
46+
from components.java
47+
artifactId = archivesBaseName
48+
pom {
49+
name = 'Durable Task Export History for Java'
50+
description = 'This package provides orchestration history export capabilities for the Durable Task Java SDK.'
51+
url = "https://github.com/microsoft/durabletask-java/tree/main/exporthistory"
52+
licenses {
53+
license {
54+
name = "MIT License"
55+
url = "https://opensource.org/licenses/MIT"
56+
distribution = "repo"
57+
}
58+
}
59+
developers {
60+
developer {
61+
id = "Microsoft"
62+
name = "Microsoft Corporation"
63+
}
64+
}
65+
scm {
66+
connection = "scm:git:https://github.com/microsoft/durabletask-java"
67+
developerConnection = "scm:git:git@github.com:microsoft/durabletask-java"
68+
url = "https://github.com/microsoft/durabletask-java/tree/main/exporthistory"
69+
}
70+
// Include compile-only dependencies in generated POM
71+
withXml {
72+
project.configurations.compileOnly.allDependencies.each { dependency ->
73+
asNode().dependencies[0].appendNode("dependency").with {
74+
it.appendNode("groupId", dependency.group)
75+
it.appendNode("artifactId", dependency.name)
76+
it.appendNode("version", dependency.version)
77+
it.appendNode("scope", "provided")
78+
}
79+
}
80+
}
81+
}
82+
}
83+
}
84+
}
85+
86+
signing {
87+
required = !project.hasProperty("skipSigning")
88+
sign publishing.publications.mavenJava
89+
}
90+
91+
java {
92+
withSourcesJar()
93+
withJavadocJar()
94+
}
95+
96+
spotbugs {
97+
toolVersion = '4.9.8'
98+
effort = com.github.spotbugs.snom.Effort.valueOf('MAX')
99+
reportLevel = com.github.spotbugs.snom.Confidence.valueOf('HIGH')
100+
ignoreFailures = true
101+
excludeFilter = file('spotbugs-exclude.xml')
102+
}
103+
104+
spotbugsMain {
105+
reports {
106+
html {
107+
required = true
108+
stylesheet = 'fancy-hist.xsl'
109+
}
110+
xml {
111+
required = true
112+
}
113+
}
114+
}
115+
116+
spotbugsTest {
117+
reports {
118+
html {
119+
required = true
120+
stylesheet = 'fancy-hist.xsl'
121+
}
122+
xml {
123+
required = true
124+
}
125+
}
126+
}

‎exporthistory/spotbugs-exclude.xml‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<FindBugsFilter>
3+
<!-- Exclude test classes -->
4+
<Match>
5+
<Class name="~.*Test"/>
6+
</Match>
7+
8+
<!-- Exclude common false positives -->
9+
<Match>
10+
<BugPattern name="DM_CONVERT_CASE"/>
11+
</Match>
12+
13+
<!-- Exclude serialization related warnings -->
14+
<Match>
15+
<BugPattern name="SE_NO_SERIALVERSIONID"/>
16+
</Match>
17+
</FindBugsFilter>

‎exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/activities/ExportInstanceHistoryActivity.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,9 +109,15 @@ public Object run(TaskActivityContext ctx) {
109109
logger.info("Successfully exported instance " + instanceId + " to " + blobPath);
110110
return new ExportResult(instanceId, true, null);
111111

112+
} catch (RuntimeException ex) {
113+
// Let transient failures (blob upload, serialization, network) propagate
114+
// so the activity-level retry policy in the orchestrator can apply.
115+
logger.log(Level.SEVERE, "Failed to export instance " + instanceId, ex);
116+
throw ex;
112117
} catch (Exception ex) {
113118
logger.log(Level.SEVERE, "Failed to export instance " + instanceId, ex);
114-
return new ExportResult(instanceId, false, ex.getMessage());
119+
throw new RuntimeException(
120+
"Export failed for instance " + instanceId + ": " + ex.getMessage(), ex);
115121
}
116122
}
117123

‎exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/client/DefaultExportHistoryClient.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,14 @@ public ExportJobQueryPageable listJobs(@Nullable ExportJobQuery filter) {
8989
if (query != null && query.getStatus() != null && state.getStatus() != query.getStatus()) {
9090
continue;
9191
}
92+
if (query != null && query.getCreatedFrom() != null
93+
&& (state.getCreatedAt() == null || state.getCreatedAt().isBefore(query.getCreatedFrom()))) {
94+
continue;
95+
}
96+
if (query != null && query.getCreatedTo() != null
97+
&& (state.getCreatedAt() == null || state.getCreatedAt().isAfter(query.getCreatedTo()))) {
98+
continue;
99+
}
92100
ExportJobDescription desc = new ExportJobDescription();
93101
desc.setJobId(metadata.getEntityInstanceId().getKey());
94102
desc.setStatus(state.getStatus());

‎exporthistory/src/main/java/com/microsoft/durabletask/exporthistory/models/ExportJobQuery.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,17 @@
77

88
/**
99
* Query filter for listing export jobs.
10+
* <p>
11+
* Filtering semantics:
12+
* <ul>
13+
* <li>{@code status} and {@code createdFrom}/{@code createdTo} are applied <em>after</em>
14+
* the underlying entity query, so a single page may contain fewer than {@link #getPageSize()}
15+
* results even when more matching jobs exist on subsequent pages. Continue paginating
16+
* using the returned continuation token to retrieve all matches.</li>
17+
* <li>The time window is inclusive on both ends: a job whose {@code createdAt} equals
18+
* {@code createdFrom} or {@code createdTo} <em>is</em> included.</li>
19+
* <li>Jobs with a {@code null} {@code createdAt} are excluded when either time filter is set.</li>
20+
* </ul>
1021
*/
1122
public final class ExportJobQuery {
1223

0 commit comments

Comments
 (0)