[Feature] Support clearing all consumers execept specified consumer in the consumer directory

[chenxinwei]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Sometimes，we may want to modify the consumerId in a Flink task.However, if we do not delete other consumers that are not actually used at this time, the expired data will not be effectively cleaned up.So I want to support Support clearing all consumers execept specified consumer in the consumer directory.

Solution

I will introduce the ClearConsumersProcedure to support clearing unspecified consumers, clearing specified consumers, and clearing all consumers in a table.

Clear consumers procedure. Usage:

1. clear all consumers except the specified consumer in the table
   CALL sys.clear_consumers('tableId', 'consumerIds', true)
2. clear all specified consumers in the table
   CALL sys.clear_consumers('tableId', 'consumerIds') or CALL sys.clear_consumers('tableId', 'consumerIds', false)
3. clear all consumers in the table
   CALL sys.clear_consumers('tableId')

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[ChaomingZhangCN]

Maybe we can use a Procedure to do this. We already have consumer.expiration-time, adding too many options reduces usability. What do you think? @chenxinwei

[chenxinwei]

Maybe we can use a Procedure to do this. We already have consumer.expiration-time, adding too many options reduces usability. What do you think? @chenxinwei

The existing consumer.expiration-time parameter causes the consumerId in use to join the expiration process. In some cases, we don't want the consumer to expire.

[ChaomingZhangCN]

I think, clearing a specified consumer by flink Procedure is a better way. @chenxinwei

CALL sys.clear_consumer('tableId', 'consumerId')

[chenxinwei]

I think, clearing a specified consumer by flink Procedure is a better way. @chenxinwei

CALL sys.clear_consumer('tableId', 'consumerId')

Thanks for your suggestion, this approach is indeed better.