fix failure events posting to kylo ops manager
@@ -85,6 +85,7 @@ public class ProvenanceFeedLookup {
}
public KyloProcessorFlowType setProcessorFlowType(ProvenanceEventRecordDTO event) {
+ if (event.getProcessorType() == null) {
Map<String, KyloProcessorFlowType> flowTypes = getFlowCache().getProcessorFlowTypesAsMap(event.getFeedName(), event.getComponentId());
@@ -99,9 +100,12 @@ public class ProvenanceFeedLookup {
type = failureType;
} else if (allType != null) {
type = allType;
- } else if (!getFlowCache().hasProcessorFlowTypesMapped(event.getFeedName())) {
- //If no processors are mapped for this feed and we got here via a failure event, then mark it as a failure
- type = failureType;
+ } else if (successType != null) {
+ type = successType;
+ }
+ // failure events need to be explicitly defined as Warnings in the template registration
+ if (type.equals(KyloProcessorFlowType.NORMAL_FLOW)) {
+ type = KyloProcessorFlowType.FAILURE;
}
} else {
@@ -113,7 +117,8 @@ public class ProvenanceFeedLookup {
}
event.setProcessorType(type);
log.debug("Setting the Flow Type as {} for Processor {} ({}) on Feed {}. Flow Types: {} ", type, event.getComponentName(), event.getComponentId(), event.getFeedName(), flowTypes);
- return type;
+ }
+ return event.getProcessorType();
}
public boolean isFailureEvent(ProvenanceEventRecordDTO eventRecordDTO) {
