Flink: Maintenance - MonitorSource

[pvary]

Implements a monitor source which emits TableChange events based on the new commits to the Iceberg table.
This will be used for Maintenance Task scheduling as described in the design doc. The Scheduling paragraph describes the require parameters of the TableChange object. Also the Monitor paragraph adds some more details.

Implements #10300

[rodmeneses]

Please link the design doc in the PR description

[rodmeneses]

some minor comments, otherwise LGTM

[stevenzwu]

do we need to pass in a @TempDir that can auto clean up files/dirs?

[pvary]

In Junit5 mixing extensions with @TempDir is not supported.
See: junit-team/junit5#1786

I found the proposed solution much more ugly than creating a temp dir using the java api

[stevenzwu]

My concern is that the temp dir (and files under it) wasn't cleaned up at the end of the test. at least, we should consider deleteOnExit()

https://stackoverflow.com/questions/15022219/does-files-createtempdirectory-remove-the-directory-after-jvm-exits-normally

[pvary]

It is taken care in the afterEach method:

Files.walk(warehouse).sorted(Comparator.reverseOrder()).map(Path::toFile).forEach(File::delete);

[stevenzwu]

ah. I missed the afterEach part. it might still make sense to add deleteOnExit as fallback, in case afterEach method throws an exception before the file cleanup step.

[pvary]

deleteOnExit will only work if the directory is empty. See: https://docs.oracle.com/javase/7/docs/api/java/io/File.html#delete%28%29

I don't think it would help much adding this to the code.

[stevenzwu]

LGTM. have one minor comment and will leave the decision to you on that one

[pvary]

Merged to main.
Thanks for the review @rodmeneses and @stevenzwu!