[Bug][Connector-V2][Assert] AssertSinkWriter supports multiple concurrent operations

[zhangshenghang]

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

The current AssertSinkWriter has a bug in the Close method to verify whether the data has been run completely, especially when dealing with multiple concurrent instances.
For example, if there are two AssertSinkWriter threads A and B, and A has finished running while B has not, A will execute the close method to count whether the overall execution is complete.

The correct logic is to perform validation after all threads have completed running, or alternatively, thread A only verifies the data it runs.

• link [fix][connector-v2][connector-assert] Optimize Assert Sink verification method #8356

Usage Scenario

No response

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[zhangshenghang]

You can use this configuration file to reproduce this issue

env {
  job.mode = "BATCH"
}

source {
  FakeSource {
    parallelism = 1
    tables_configs = [
      {
        row.num = 1000
        schema = {
          table = "test.abc"
          columns = [
            {
              name = "id"
              type = "bigint"
            },
            {
              name = "name"
              type = "string"
            },
            {
              name = "age"
              type = "int"
            }
          ]
        }
      },
      {
        row.num = 1000
        schema = {
          table = "test.xyz"
          columns = [
            {
              name = "id"
              type = "bigint"
            },
            {
              name = "name"
              type = "string"
            },
            {
              name = "age"
              type = "int"
            }
          ]
        }
      },
      {
        row.num = 1000
        schema = {
          table = "test.www"
          columns = [
            {
              name = "id"
              type = "bigint"
            },
            {
              name = "name"
              type = "string"
            },
            {
              name = "age"
              type = "int"
            }
          ]
        }
      }
    ]
  }
}

transform {
  FilterRowKind {
    table_match_regex = "test.a.*"
    table_transform = [{
      table_path = "test.xyz"
      exclude_kinds = ["INSERT"]
    }]
    exclude_kinds = ["INSERT"]
  }
}

sink {
  Assert {
    rules =
      {
        tables_configs = [
          {
            table_path = "test.abc"
            row_rules = [
              {
                rule_type = MIN_ROW
                rule_value = 0
              },
              {
                rule_type = MAX_ROW
                rule_value = 0
              }
            ]
          },
          {
            table_path = "test.xyz"
            row_rules = [
              {
                rule_type = MIN_ROW
                rule_value = 0
              },
              {
                rule_type = MAX_ROW
                rule_value = 0
              }
            ]
          },
          {
            table_path = "test.www"
            row_rules = [
              {
                rule_type = MIN_ROW
                rule_value = 1000
              },
              {
                rule_type = MAX_ROW
                rule_value = 1000
              }
            ]
          }
        ]
      }
  }
}

[corgy-w]

I'm in 'rehabilitation training' [/dog], if I have time I can try this issue