[FLINK-35346][table-common] Introduce workflow scheduler interface for materialized table #24767
Conversation
lsyldliu
force-pushed
the
workflow_interface
branch
3 times, most recently
from
befaab5
to
b58749b
Compare
lsyldliu
changed the title
lsyldliu
changed the title
| * Close this workflow scheduler when it is no longer needed and release any resource that it | ||
| * might be holding. | ||
| * | ||
| * @throws WorkflowException if close the related resources of workflow scheduler failed |
There was a problem hiding this comment.
| * @throws WorkflowException if close the related resources of workflow scheduler failed | |
| * @throws WorkflowException if closing the related resources of workflow scheduler failed |
| * @param createRefreshWorkflow The detail info for create refresh workflow of materialized | ||
| * table. | ||
| * @return The meta info which points to the refresh workflow in scheduler service. | ||
| * @throws WorkflowException if create refresh workflow failed |
There was a problem hiding this comment.
| * @throws WorkflowException if create refresh workflow failed | |
| * @throws WorkflowException if creating refresh workflow failed |
| * @param createRefreshWorkflow The detail info for create refresh workflow of materialized | ||
| * table. | ||
| * @return The meta info which points to the refresh workflow in scheduler service. | ||
| * @throws WorkflowException if create refresh workflow failed |
There was a problem hiding this comment.
As I understand it, there may be two possible implementations for the current CreateRefreshWorkflow in the future, corn and one-time. If there is a WorkFlowScheduler that supports only one of these two types, should we throw an unsupported exception at this point?
There was a problem hiding this comment.
I think it is no need, you can the related interface methods of Catalog, it doesn't thrown the unsupported exception in method signature even though some operations unsupported. Moreover, we have introduced the WorkflowException for WorkflowScheduler, I think we should try to wrap the exception to WorkflowExeception, it is checked Exception, user should sense it.
| * because it is optional for materialized table. | ||
| */ | ||
| public static @Nullable WorkflowScheduler<?> createWorkflowScheduler( | ||
| Configuration configuration, ClassLoader classLoader) { |
There was a problem hiding this comment.
Is it possible to pass in a ReadableConfig here? Such as TableConfig? If so, we can directly use the following method to get all options.
ConfigOption<Map<String, String>> workflowSchedulerOptions = new
ConfigOptions.key(WORKFLOW_SCHEDULER_PREFIX+ "." + identifier)
.mapType()
.defaultValue(new HashMap<>());
Map<String, String> options = configuration.get(workflowSchedulerOptions);
There was a problem hiding this comment.
No, MaterializedTableManager will be created when open a session, it will be as a member of SessionState, we can only get the Configuration on construct SessionState.
lsyldliu
force-pushed
the
workflow_interface
branch
from
b58749b
to
7c9d865
Compare
There was a problem hiding this comment.
@hackergin Thanks for your reviewing, I've addressed your comments.
|
@lsyldliu Thanks for the updates and explanations, LGTM. |
What is the purpose of the change
Introduce workflow scheduler interface for materialized table
Brief change log
Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (yes)Documentation