Flink: Maintenance - TriggerManager

[pvary]

The responsibility of the Trigger Manager is to start the Maintenance Tasks based on the incoming Table Change messages and prevent overlapping Maintenance Task runs. The event time of the Trigger messages should be monotonically increasing, as this will be used as a watermark to separate Maintenance Task runs.

Implementation for #10301

This will be used for Maintenance Trigger Manager as described in the design doc. The Trigger Manager paragraph describes the requirements in more detail.

[pvary]

Hi @stevenzwu,

I would like to move forward with this PR.
I see the following open questions:

1. Lock implementation - I still think it is important to provide a working implementation without adding a requirement for external systems. For this the Tag based solution is working for every Iceberg deployment and there were no objections or better suggestions for it during the discussion on the mailing list (https://lists.apache.org/thread/vjf8m5wg840o58yz4y3q35k2mfhbm49l). Also tried to ask help from the Flink mailing list for alternative solutions, but I got a single one offline which turned out a dead end (https://lists.apache.org/thread/3ss83w1hp8wpocx13kcqr7qyjnghvnc8). - I propose to remove the JVM based lock, and move forward with the Tag based lock
2. TableChange parameters - I am open to discussing the list of parameters which will drive the triggers. It will touch a big part of the previous PR as well, and will make this change even harder to review/read. - I propose to do it in the next PR
3. There were some questions about the overall architecture. This is described in the original proposal doc here. - I'm open to discuss the proposal offline too, if that is needed.

WDYT @stevenzwu? Could we move forward along these lines, or do you have another proposal? Do you have any foundational concerns about the PR?

Thanks,
Peter

[singhpk234]

Why are creating new interfaces rather than enhancing the existing ones ?
https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/LockManager.java

[pvary]

Notice the different requirements for the methods. This lock is not reentrant, and the unlock removes locks held by others.

[pvary]

Merged to master.
Thanks for the review @stevenzwu, @singhpk234 and @gyfora!

[tedyu]

With SQLException caught on line 257, it seems this clause is not reachable.
For instanceId(), SQLException is caught inside the method as well.

[pvary]

The pool.run(..) method could throw SQLException. For example on reconnect.

[tedyu]

I looked at the reconnect method in core/src/main/java/org/apache/iceberg/jdbc/JdbcClientPool.java.

It seems SQLException is caught there as well.

BTW thanks for the quick response.

[pvary]

The JdbcClientPool extends ClientPoolImpl<Connection, SQLException>. The ClientPoolImpl defines run like this:

  @Override
  public <R> R run(Action<R, C, E> action) throws E, InterruptedException {
    return run(action, retryByDefault);
  }

So we either need to declare SQLException in throws, or we need to handle the exception.

[tedyu]

no be null -> not be null