[INLONG-11508][Manager] Add APIs to dirty data query (#11509)

* [INLONG-11508][Manager] Add APIs to dirty data query

inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/DirtyQueryLogEntity.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.dao.entity;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+@Data
+public class DirtyQueryLogEntity implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private Integer id;
+    private String md5;
+    private String requestParams;
+    private String taskId;
+    private Integer isDeleted;
+    private String creator;
+    private String modifier;
+    private Date createTime;
+    private Date modifyTime;
+    private Integer version;
+}

inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/DirtyQueryLogEntityMapper.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.dao.mapper;
+
+import org.apache.inlong.manager.dao.entity.DirtyQueryLogEntity;
+
+import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface DirtyQueryLogEntityMapper {
+
+    int updateByIdSelective(DirtyQueryLogEntity record);
+
+    int insert(DirtyQueryLogEntity record);
+
+    DirtyQueryLogEntity selectByPrimaryKey(Integer id);
+
+    DirtyQueryLogEntity selectByMd5(String md5);
+
+    void updateToTimeout(@Param("beforeMinutes") Integer beforeMinutes);
+
+}

inlong-manager/manager-dao/src/main/resources/mappers/DirtyQueryLogEntityMapper.xml

@@ -0,0 +1,93 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements. See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership. The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License. You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied. See the License for the
+    specific language governing permissions and limitations
+    under the License.
+-->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="org.apache.inlong.manager.dao.mapper.DirtyQueryLogEntityMapper">
+    <resultMap id="BaseResultMap" type="org.apache.inlong.manager.dao.entity.DirtyQueryLogEntity">
+        <id column="id" jdbcType="INTEGER" property="id" />
+        <result column="md5" jdbcType="VARCHAR" property="md5" />
+        <result column="request_params" jdbcType="VARCHAR" property="requestParams" />
+        <result column="task_id" jdbcType="VARCHAR" property="taskId" />
+        <result column="is_deleted" jdbcType="INTEGER" property="isDeleted" />
+        <result column="creator" jdbcType="VARCHAR" property="creator" />
+        <result column="modifier" jdbcType="VARCHAR" property="modifier" />
+        <result column="create_time" jdbcType="TIMESTAMP" property="createTime" />
+        <result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime" />
+        <result column="version" jdbcType="INTEGER" property="version" />
+    </resultMap>
+
+    <sql id="Base_Column_List">
+        id, md5, request_params, task_id, is_deleted, creator, modifier, create_time, modify_time, version
+    </sql>
+    <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
+        select
+        <include refid="Base_Column_List" />
+        from dirty_query_log
+        where is_deleted = 0
+        and id = #{id,jdbcType=INTEGER}
+    </select>
+    <select id="selectByMd5" parameterType="java.lang.String" resultMap="BaseResultMap">
+        select
+        <include refid="Base_Column_List" />
+        from dirty_query_log
+        where is_deleted = 0
+        and md5 = #{md5,jdbcType=VARCHAR}
+        limit 1
+    </select>
+    <insert id="insert" parameterType="org.apache.inlong.manager.dao.entity.DirtyQueryLogEntity">
+        insert into dirty_query_log (id, md5, request_params,
+                                     task_id, creator, modifier)
+        values (#{id,jdbcType=INTEGER}, #{md5,jdbcType=VARCHAR}, #{requestParams,jdbcType=VARCHAR},
+                #{taskId,jdbcType=VARCHAR}, #{creator,jdbcType=VARCHAR}, #{modifier,jdbcType=VARCHAR})
+    </insert>
+    <update id="updateByIdSelective" parameterType="org.apache.inlong.manager.dao.entity.DirtyQueryLogEntity">
+        update dirty_query_log
+        <set>
+            <if test="md5 != null">
+                md5 = #{md5,jdbcType=VARCHAR},
+            </if>
+            <if test="requestParams != null">
+                request_params = #{requestParams,jdbcType=VARCHAR},
+            </if>
+            <if test="taskId != null">
+                task_id = #{taskId,jdbcType=VARCHAR},
+            </if>
+            <if test="isDeleted != null">
+                is_deleted = #{isDeleted,jdbcType=INTEGER},
+            </if>
+            <if test="modifier != null">
+                modifier = #{modifier,jdbcType=VARCHAR},
+            </if>
+            version = #{version,jdbcType=INTEGER} + 1
+        </set>
+        where id = #{id,jdbcType=INTEGER}
+        and version = #{version,jdbcType=INTEGER}
+    </update>
+    <update id="updateToTimeout">
+        update dirty_query_log
+        <set>
+            is_deleted = id
+        </set>
+        <where>
+            is_deleted = 0
+            and create_time &lt;= DATE_ADD(NOW(), INTERVAL -#{beforeMinutes, jdbcType=INTEGER} MINUTE)
+        </where>
+    </update>
+</mapper>

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/BaseStreamSink.java

@@ -17,12 +17,18 @@
 
 package org.apache.inlong.manager.pojo.sink;
 
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.JsonUtils;
+
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 
+import javax.validation.constraints.NotNull;
+
 /**
  * The base parameter class of StreamSink, support user extend their own business params.
  */
@@ -32,6 +38,9 @@
 @ApiModel("Base info of stream sink")
 public class BaseStreamSink {
 
+    @ApiModelProperty("Enable data archiving")
+    private Boolean enableDataArchiving;
+
     @ApiModelProperty("Transform sql")
     private String transformSql;
 
@@ -40,4 +49,12 @@ public class BaseStreamSink {
 
     @ApiModelProperty("Stop consume time, yyyy-MM-dd HH:mm:ss format")
     private String stopConsumeTime;
+
+    public static BaseStreamSink getFromJson(@NotNull String extParams) {
+        try {
+            return JsonUtils.parseObject(extParams, BaseStreamSink.class);
+        } catch (Exception e) {
+            throw new BusinessException(ErrorCodeEnum.SINK_INFO_INCORRECT.getMessage() + ": " + e.getMessage());
+        }
+    }
 }

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/DirtyDataDetailResponse.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.sink;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+/**
+ * Dirty data detail info.
+ */
+@Data
+@ApiModel("Dirty data detail info")
+public class DirtyDataDetailResponse {
+
+    @ApiModelProperty(value = "Tdbank imp date")
+    private String tdbankImpDate;
+
+    @ApiModelProperty(value = "Data flow id")
+    private String dataFlowId;
+
+    @ApiModelProperty(value = "Inlong group id")
+    private String groupId;
+
+    @ApiModelProperty(value = "Inlong stream id")
+    private String streamId;
+
+    @ApiModelProperty(value = "Report time")
+    private String reportTime;
+
+    @ApiModelProperty(value = "Data time")
+    private String dataTime;
+
+    @ApiModelProperty(value = "Server type")
+    private String serverType;
+
+    @ApiModelProperty(value = "Dirty type")
+    private String dirtyType;
+
+    @ApiModelProperty(value = "Dirty message")
+    private String dirtyMessage;
+
+    @ApiModelProperty(value = "Ext info")
+    private String extInfo;
+
+    @ApiModelProperty(value = "Dirty data")
+    private String dirtyData;
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/DirtyDataRequest.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.sink;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.util.List;
+
+/**
+ * Query request for Dirty data
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@ApiModel("Query request for Dirty data")
+public class DirtyDataRequest {
+
+    @ApiModelProperty(value = "Sink id list")
+    private List<Integer> sinkIdList;
+
+    @ApiModelProperty(value = "Key word")
+    private String keyword;
+
+    @ApiModelProperty(value = "Server type")
+    private String serverType;
+
+    @ApiModelProperty(value = "Dirty type")
+    private String dirtyType;
+
+    @ApiModelProperty(value = "Start time")
+    private String startTime;
+
+    @ApiModelProperty(value = "End time")
+    private String endTime;
+
+    @ApiModelProperty(value = "Data count")
+    private Integer dataCount = 10;
+}

inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sink/DirtyDataResponse.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.inlong.manager.pojo.sink;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+/**
+ * Dirty data info.
+ */
+@Data
+@ApiModel("Dirty data info")
+public class DirtyDataResponse {
+
+    @ApiModelProperty(value = "Task id")
+    private String taskId;
+
+}