[Fix][connector-rocketmq] commit a correct offset to broker & reduce ThreadInterruptedException log

[YalikWang]

Purpose of this pull request

Fix issue 6645 and 6665.
Reduce ThreadInterruptedException log print on the console.
Commit a correct offset to the broker when snapshot.

Does this PR introduce any user-facing change?

No

How was this patch tested?

test locally and test with existed test cases

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update change log that in connector document. For more details you can refer to connector-v2
  2. Update plugin-mapping.properties and add new connector information in it
  3. Update the pom file of seatunnel-dist
• Update the release-note.

[Hisoka-X]

Thanks @YalikWang for created this PR! Is it possible to add some test cases for these bugs?

[YalikWang]

Thanks @YalikWang for created this PR! Is it possible to add some test cases for these bugs?

Yes, I am willing to add some test cases. e2e or unit test ？Do you have any additional suggestions about it ?

[Hisoka-X]

Thanks @YalikWang for created this PR! Is it possible to add some test cases for these bugs?

Yes, I am willing to add some test cases. e2e or unit test ？Do you have any additional suggestions about it ?

E2E would be better. Because we should check offset in rocketmq after connector read data, this is only way to make sure same bug not happen again.

[Hisoka-X]

The CI failed. Could you fix it?

[YalikWang]

The CI failed. Could you fix it?

I am trying. The reason for this failure is that I did not set up the correct consumer.group. I fixed it today, but the problem still exists： In E2E testing, when the batch job ends, the consume offset cannot be read through RocketmqAdminApi. I hope it can be resolved tomorrow。

[YalikWang]

@Hisoka-X
I found that,in e2e testing, rocketmq source connector won't submit consume offset when batch job finished with flink13. I did not test with zeta e2e because of env error.
This is my job conf:

env {
  parallelism = 1
  job.mode = "BATCH"
  checkpoint.interval = 1000

  # You can set spark configuration here
  spark.app.name = "SeaTunnel"
  spark.executor.instances = 1
  spark.executor.cores = 1
  spark.executor.memory = "1g"
  spark.master = local
}

source {
  Rocketmq {
    name.srv.addr = "rocketmq-e2e:9876"
    topics = "test_topic_text_offset"
    result_table_name = "rocketmq_table"
    schema = {
      fields {
        id = bigint
        c_map = "map<string, smallint>"
        c_array = "array<tinyint>"
        c_string = string
        c_boolean = boolean
        c_tinyint = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_decimal = "decimal(2, 1)"
        c_bytes = bytes
        c_date = date
        c_timestamp = timestamp
      }
    }
    format = text
    # The default field delimiter is ","
    field_delimiter = ","
  }
}

transform {
}

sink {
  Console{
    source_table_name = "rocketmq_table"
  }
}

And this is the consumerOffset.json in rocektmq broker container (4.9.4):

 cat consumerOffset.json
{
        "offsetTable":{
                "test_topic_text_offset@SeaTunnel-Consumer-Group":{0:0,1:0,2:0,3:0
                }
        }
}

This is topic status in rocektmq broker container (4.9.4):
./mqadmin topicStatus -n localhost:9876 -t test_topic_text_offset

#Broker Name	#QID #Min Offset	#Max Offset	#Last Updated
broker-a	0	0	10
broker-a	1	0	0
broker-a	2	0	0
broker-a	3	0	0

If I call the function to submit offset when there is no more element (RocketMqSourceReader#pollNext) or reader close (RocketMqSourceReader#pollNext), it does work. But I'm not sure if this modification is appropriate. Could you give me some advice about it ?

[Hisoka-X]

I found that,in e2e testing, rocketmq source connector won't submit consume offset when batch job finished with flink13.

In batch mode, this is a known issue, we can skip it for now. It should work fine in zeta

If I call the function to submit offset when there is no more element (RocketMqSourceReader#pollNext) or reader close (RocketMqSourceReader#pollNext), it does work. But I'm not sure if this modification is appropriate. Could you give me some advice about it ?

Offset only can be submited in notifyCheckpointComplete.

[YalikWang]

I have rolled back the submission regarding rocketmq consume offset verification in the e2e test, and now the CI has passed. Perhaps in the future, we can add this check in e2e test.

[Hisoka-X]

I have rolled back the submission regarding rocketmq consume offset verification in the e2e test, and now the CI has passed. Perhaps in the future, we can add this check in e2e test.

How about only verify it in the zeta?

[YalikWang]

I have rolled back the submission regarding rocketmq consume offset verification in the e2e test, and now the CI has passed. Perhaps in the future, we can add this check in e2e test.

How about only verify it in the zeta?

I added an e2e test case to check the consumption offset, which only runs with zeta.

[Hisoka-X]

Over all LGTM.