You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "pratapaditya04 (via GitHub)" <gi...@apache.org> on 2024/09/11 07:02:09 UTC
[PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
pratapaditya04 opened a new pull request, #4052:
URL: https://github.com/apache/gobblin/pull/4052
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-2153
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
Added flowGroup.flowName and user.to.proxy as SearchAttributes to filter Temporal Flows in the UI
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Blazer-007 (via GitHub)" <gi...@apache.org>.
Blazer-007 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756762437
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
I was thinking of Optional<> as return type, since it is null safe by default but as a parameter can be good way too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1760528729
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
Don't think returning Optional is needed, as we are returning an empty Map instead of null.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756459307
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
Review Comment:
have added default values so incase FLOW_GROUP_KEY, FLOW_NAME_KEY and USER_TO_PROXY_KEY are not set, the default values will be returned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756459307
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
Review Comment:
have added default values so incase FLOW_GROUP_KEY, FLOW_NAME_KEY and USER_TO_PROXY_KEY are not set, the default values will be returned. Dont prefer throwing an exception as we don't want the job to fail just because searchAttributes couldn't be set.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1757366881
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -57,21 +59,25 @@ public CommitStats process(WUProcessingSpec workSpec) {
}
private CommitStats performWork(WUProcessingSpec workSpec) {
- Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
- NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec);
- int workunitsProcessed = processingWorkflow.performWorkload(
- WorkflowAddr.ROOT, workload, 0,
- workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty()
- );
- if (workunitsProcessed > 0) {
- CommitStepWorkflow commitWorkflow = createCommitStepWorkflow();
- CommitStats result = commitWorkflow.commit(workSpec);
- if (result.getNumCommittedWorkUnits() == 0) {
- log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ try {
+ Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
+ JobState jobState = Help.loadJobState(workSpec, Help.loadFileSystem(workSpec));
+ NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec, jobState);
+ int workunitsProcessed = processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ if (workunitsProcessed > 0) {
+ CommitStepWorkflow commitWorkflow = createCommitStepWorkflow(jobState);
+ CommitStats result = commitWorkflow.commit(workSpec);
+ if (result.getNumCommittedWorkUnits() == 0) {
+ log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ }
+ return result;
+ } else {
+ log.error("No work units processed, so no commit attempted.");
+ return CommitStats.createEmpty();
}
- return result;
- } else {
- log.error("No work units processed, so no commit attempted.");
+ } catch (Exception ignored) {
+ log.error("Exception occured during performing Work", ignored);
return CommitStats.createEmpty();
Review Comment:
If it throws an exception when loading the jobState (which I presume would lead to an empty jobState) we should just throw an ApplicationException in this scenario, so that we can push the Temporal framework to retry the workflow rather than progressing silently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1760530246
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -57,21 +59,25 @@ public CommitStats process(WUProcessingSpec workSpec) {
}
private CommitStats performWork(WUProcessingSpec workSpec) {
- Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
- NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec);
- int workunitsProcessed = processingWorkflow.performWorkload(
- WorkflowAddr.ROOT, workload, 0,
- workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty()
- );
- if (workunitsProcessed > 0) {
- CommitStepWorkflow commitWorkflow = createCommitStepWorkflow();
- CommitStats result = commitWorkflow.commit(workSpec);
- if (result.getNumCommittedWorkUnits() == 0) {
- log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ try {
+ Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
+ JobState jobState = Help.loadJobState(workSpec, Help.loadFileSystem(workSpec));
+ NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec, jobState);
+ int workunitsProcessed = processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ if (workunitsProcessed > 0) {
+ CommitStepWorkflow commitWorkflow = createCommitStepWorkflow(jobState);
+ CommitStats result = commitWorkflow.commit(workSpec);
+ if (result.getNumCommittedWorkUnits() == 0) {
+ log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ }
+ return result;
+ } else {
+ log.error("No work units processed, so no commit attempted.");
+ return CommitStats.createEmpty();
}
- return result;
- } else {
- log.error("No work units processed, so no commit attempted.");
+ } catch (Exception ignored) {
+ log.error("Exception occured during performing Work", ignored);
return CommitStats.createEmpty();
Review Comment:
makes sense, addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Blazer-007 (via GitHub)" <gi...@apache.org>.
Blazer-007 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756762437
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
Optional<> as return type, since it is null safe by default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo merged PR #4052:
URL: https://github.com/apache/gobblin/pull/4052
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "khandelwal-prateek (via GitHub)" <gi...@apache.org>.
khandelwal-prateek commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1755105657
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
javadoc mentions that the method should return an empty map if searchAttributes is null, however, this returns null. We can use `Collections.emptyMap()` here instead of returning null
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
Review Comment:
let's add null checks for required keys in jobProps. If `FLOW_GROUP_KEY`, `FLOW_NAME_KEY`, or `USER_TO_PROXY_KEY` are missing, the method will currently throw a NullPointerException. It would be better to handle this gracefully by throwing an IllegalArgumentException with an error message.
##########
gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtilsTest.java:
##########
@@ -0,0 +1,78 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class TemporalWorkFlowUtilsTest {
+
+ private Properties jobProps = Mockito.mock(Properties.class);
+
+ @Test
+ public void testGenerateGaasSearchAttributes() {
+ // Mocking the properties
+ Mockito.when(jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY)).thenReturn("group");
+ Mockito.when(jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)).thenReturn("name");
+ Mockito.when(jobProps.getProperty(Help.USER_TO_PROXY_KEY)).thenReturn("userProxy");
Review Comment:
the test assumes that the required properties(`FLOW_GROUP_KEY`, `FLOW_NAME_KEY` and `USER_TO_PROXY_KEY`) are always present. It would be useful to add a test where some of these properties are missing or null and test the behaviour in such cases
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Blazer-007 (via GitHub)" <gi...@apache.org>.
Blazer-007 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1755172800
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
l believe using Optional<> can be more intuitive in such cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756463470
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
+ }
+ Map<String, Object> convertedAttributes = new HashMap<>();
+
+ convertedAttributes.putAll(searchAttributes);
Review Comment:
simplified the conversion, however have kept this function as we need validations as new HashMap<>(searchAttributes) throws NPE in case passed parameter is null
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
Review Comment:
addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1755945665
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
Review Comment:
we should call this GAAS_FLOW_ID_SEARCH_KEY, as flowGroup + flowName is the UUID for a given flow config in GaaS
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -57,21 +59,25 @@ public CommitStats process(WUProcessingSpec workSpec) {
}
private CommitStats performWork(WUProcessingSpec workSpec) {
- Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
- NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec);
- int workunitsProcessed = processingWorkflow.performWorkload(
- WorkflowAddr.ROOT, workload, 0,
- workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty()
- );
- if (workunitsProcessed > 0) {
- CommitStepWorkflow commitWorkflow = createCommitStepWorkflow();
- CommitStats result = commitWorkflow.commit(workSpec);
- if (result.getNumCommittedWorkUnits() == 0) {
- log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ try {
+ Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
+ JobState jobState = Help.loadJobState(workSpec, Help.loadFileSystem(workSpec));
+ NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec, jobState);
+ int workunitsProcessed = processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ if (workunitsProcessed > 0) {
+ CommitStepWorkflow commitWorkflow = createCommitStepWorkflow(jobState);
+ CommitStats result = commitWorkflow.commit(workSpec);
+ if (result.getNumCommittedWorkUnits() == 0) {
+ log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ }
+ return result;
+ } else {
+ log.error("No work units processed, so no commit attempted.");
+ return CommitStats.createEmpty();
}
- return result;
- } else {
- log.error("No work units processed, so no commit attempted.");
+ } catch (Exception ignored) {
+ log.error("Exception occured during performing Work", ignored);
return CommitStats.createEmpty();
Review Comment:
What's the rationale for this try catch? Seems dangerous to ignore exceptions during processworkunits step, shouldn't we want it to fail loudly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "khandelwal-prateek (via GitHub)" <gi...@apache.org>.
khandelwal-prateek commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1755153831
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
+ }
+ Map<String, Object> convertedAttributes = new HashMap<>();
+
+ convertedAttributes.putAll(searchAttributes);
Review Comment:
It seems like the intention of this method is to convert the input map `Map<String, List<?>>` into a `Map<String, Object>` without transforming the list values into single objects. This method seems to be written for setSearchAttributes on ChildWorkflowOptions in AbstractNestingExecWorkflowImpl(which expects Map<String, Object>), we can simplify the implementation by directly using `new HashMap<>(searchAttributes)`. This would achieve the same result more efficiently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756461822
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtils.java:
##########
@@ -0,0 +1,52 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import lombok.NonNull;
+import lombok.experimental.UtilityClass;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+
+
+/**
+ * Utility class for handling Temporal workflow-related operations.
+ */
+@UtilityClass
+public class TemporalWorkFlowUtils {
+
+ /**
+ * Generates search attributes for a WorkFlow based on the provided GAAS job properties.
+ *
+ * @param jobProps the properties of the job, must not be null.
+ * @return a map containing the generated search attributes.
+ */
+ public static Map<String, Object> generateGaasSearchAttributes(@NonNull Properties jobProps) {
+ Map<String, Object> attributes = new HashMap<>();
+ attributes.put(Help.GAAS_FLOW_KEY, String.format("%s.%s", jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY),
+ jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)));
+ attributes.put(Help.USER_TO_PROXY_SEARCH_KEY, jobProps.getProperty(Help.USER_TO_PROXY_KEY));
+ return attributes;
+ }
+
+ /**
+ * Converts search attribute values from a map of lists to a map of objects.
+ *
+ * @param searchAttributes a map where the keys are attribute names and the values are lists of attribute values.
+ * Can be null.
+ * @return a map where the keys are attribute names and the values are the corresponding attribute values.
+ * If the input map is null, an empty map is returned.
+ */
+ public static Map<String, Object> convertSearchAttributesValuesFromListToObject(
+ Map<String, List<?>> searchAttributes) {
+ if (searchAttributes == null) {
+ return null;
Review Comment:
1. addressed.
2. Do you mean Optional<> as the parameter to avoid null check or the return type?
##########
gobblin-temporal/src/test/java/org/apache/gobblin/temporal/ddm/util/TemporalWorkFlowUtilsTest.java:
##########
@@ -0,0 +1,78 @@
+package org.apache.gobblin.temporal.ddm.util;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.temporal.ddm.work.assistance.Help;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+public class TemporalWorkFlowUtilsTest {
+
+ private Properties jobProps = Mockito.mock(Properties.class);
+
+ @Test
+ public void testGenerateGaasSearchAttributes() {
+ // Mocking the properties
+ Mockito.when(jobProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY)).thenReturn("group");
+ Mockito.when(jobProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY)).thenReturn("name");
+ Mockito.when(jobProps.getProperty(Help.USER_TO_PROXY_KEY)).thenReturn("userProxy");
Review Comment:
added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [GOBBLIN-2153]added search attributes to Temporal workflows [gobblin]
Posted by "pratapaditya04 (via GitHub)" <gi...@apache.org>.
pratapaditya04 commented on code in PR #4052:
URL: https://github.com/apache/gobblin/pull/4052#discussion_r1756465002
##########
gobblin-temporal/src/main/java/org/apache/gobblin/temporal/ddm/workflow/impl/ProcessWorkUnitsWorkflowImpl.java:
##########
@@ -57,21 +59,25 @@ public CommitStats process(WUProcessingSpec workSpec) {
}
private CommitStats performWork(WUProcessingSpec workSpec) {
- Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
- NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec);
- int workunitsProcessed = processingWorkflow.performWorkload(
- WorkflowAddr.ROOT, workload, 0,
- workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty()
- );
- if (workunitsProcessed > 0) {
- CommitStepWorkflow commitWorkflow = createCommitStepWorkflow();
- CommitStats result = commitWorkflow.commit(workSpec);
- if (result.getNumCommittedWorkUnits() == 0) {
- log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ try {
+ Workload<WorkUnitClaimCheck> workload = createWorkload(workSpec);
+ JobState jobState = Help.loadJobState(workSpec, Help.loadFileSystem(workSpec));
+ NestingExecWorkflow<WorkUnitClaimCheck> processingWorkflow = createProcessingWorkflow(workSpec, jobState);
+ int workunitsProcessed = processingWorkflow.performWorkload(WorkflowAddr.ROOT, workload, 0,
+ workSpec.getTuning().getMaxBranchesPerTree(), workSpec.getTuning().getMaxSubTreesPerTree(), Optional.empty());
+ if (workunitsProcessed > 0) {
+ CommitStepWorkflow commitWorkflow = createCommitStepWorkflow(jobState);
+ CommitStats result = commitWorkflow.commit(workSpec);
+ if (result.getNumCommittedWorkUnits() == 0) {
+ log.warn("No work units committed at the job level. They could have been committed at the task level.");
+ }
+ return result;
+ } else {
+ log.error("No work units processed, so no commit attempted.");
+ return CommitStats.createEmpty();
}
- return result;
- } else {
- log.error("No work units processed, so no commit attempted.");
+ } catch (Exception ignored) {
+ log.error("Exception occured during performing Work", ignored);
return CommitStats.createEmpty();
Review Comment:
Help.loadJobState(workSpec, Help.loadFileSystem(workSpec)) throws a checked exception which needs to be catched, however have reduced the scope of the try catch block to this line
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org