fix feedback error bug

edp963 · Mar 12, 2019 · cea0992 · cea0992
1 parent 3765f45
commit cea0992
Show file tree

Hide file tree

Showing 3 changed files with 3 additions and 3 deletions.
diff --git a/flinkx/src/main/scala/edp/wormhole/flinkx/common/FlinkxUtils.scala b/flinkx/src/main/scala/edp/wormhole/flinkx/common/FlinkxUtils.scala
@@ -58,7 +58,7 @@ object FlinkxUtils {
  streamId, DateUtils.currentDateTime, sinkNamespace, UmsWatermark(ts),
  UmsWatermark(ts), errorCount, errorMsg, batchId, null,protocolType,
  flowId,errorPattern),
- Some(UmsProtocolType.FEEDBACK_SPARKX_FLOW_ERROR + "." + streamId),
+ Some(UmsProtocolType.FEEDBACK_FLOW_ERROR + "." + flowId),
  config.kafka_output.brokers)
  }
 

diff --git a/sparkx/src/main/scala/edp/wormhole/sparkx/common/SparkxUtils.scala b/sparkx/src/main/scala/edp/wormhole/sparkx/common/SparkxUtils.scala
@@ -67,7 +67,7 @@ object SparkxUtils {
  config.spark_config.stream_id, DateUtils.currentDateTime, sinkNamespace, UmsWatermark(ts),
  UmsWatermark(ts), errorCount, errorMsg, batchId, tmpJsonArray.toJSONString,protocolType,
  flowId,errorPattern),
- Some(UmsProtocolType.FEEDBACK_SPARKX_FLOW_ERROR + "." + config.spark_config.stream_id),
+ Some(UmsProtocolType.FEEDBACK_FLOW_ERROR + "." + flowId),
  config.kafka_output.brokers)
  }
 

diff --git a/ums/src/main/scala/edp/wormhole/ums/UmsProtocolUtils.scala b/ums/src/main/scala/edp/wormhole/ums/UmsProtocolUtils.scala
@@ -215,7 +215,7 @@ trait UmsProtocolUtils {
  dataType: String,
  flowId: Long,
  errorPattern: String) = toJsonCompact(Ums(
- protocol = UmsProtocol(UmsProtocolType.FEEDBACK_SPARKX_FLOW_ERROR),
+ protocol = UmsProtocol(UmsProtocolType.FEEDBACK_FLOW_ERROR),
  schema = UmsSchema(sourceNamespace, Some(Seq(
  UmsField(UmsSysField.TS.toString, UmsFieldType.STRING),
  UmsField("sink_namespace", UmsFieldType.STRING),