Replication job to trigger setup and carbon flow for replica tables [WIP] #276
Conversation
rohitkum2506
force-pushed
the
rohikuma/ReplicationJob
branch
from
727def4 to
4ceabdd
Compare
There was a problem hiding this comment.
Thanks @rohitkum2506. Did initial pass and added some comments.
apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableReplicationTask.java
Outdated
Show resolved
Hide resolved
apps/spark/src/main/java/com/linkedin/openhouse/jobs/util/ReplicationConfig.java
Outdated
Show resolved
Hide resolved
| } | ||
| List<ReplicationConfig> replicationConfigList = new ArrayList<>(); | ||
| Replication replication = response.getPolicies().getReplication(); | ||
| List<com.linkedin.openhouse.tables.client.model.ReplicationConfig> replicationConfig = |
There was a problem hiding this comment.
So we are using two classes here with different namespace. Is the class com.linkedin.openhouse.jobs.util.ReplicationConfig internal here and returned as part of method response?
There was a problem hiding this comment.
Yes, com.linkedin.openhouse.jobs.util.ReplicationConfig is meant as translation layer between jobs model object and maintenance jobs
apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableReplicationTask.java
Outdated
Show resolved
Hide resolved
apps/spark/src/main/java/com/linkedin/openhouse/jobs/scheduler/tasks/TableReplicationTask.java
Outdated
Show resolved
Hide resolved
| .filter(m -> m.isPrimary() && (m.getReplicationConfig() != null)) | ||
| .collect(Collectors.toList()); | ||
| log.info( | ||
| "Fetched metadata for {} tables for replication setup task", |
There was a problem hiding this comment.
Nit: Better to print the table names as comma separated list.
There was a problem hiding this comment.
But the list can be more. Is there a way to track for which tables replication is being setup for every scheduled run?
There was a problem hiding this comment.
List of tables could be large. Each task will run for a table and log should capture this detail task.
| @@ -86,6 +93,31 @@ private Optional<RetentionConfig> getTableRetention(GetTableResponseBody respons | |||
| .build()); | |||
| } | |||
|
|
|||
| private Optional<List<ReplicationConfig>> getTableReplication(GetTableResponseBody response) { | |||
There was a problem hiding this comment.
How do we plan to filter tables as replication setup is one time activity? Would that be part of li repo?
- Identify the tables for which replication setup is needed like recent tables.
- Identify the tables for which replication config is updated.
There was a problem hiding this comment.
Or we can consider last updated time and include the change in this PR?
Summary
New job workflow to run Replication setup process on Airflow. Applies to primary tables with defined ReplicationConfig.
Design decisions:
Instead it will use Airflow client to trigger and manage lifecycle of a Airflow job.
Future work:
Changes
For all the boxes checked, please include additional details of the changes made in this pull request.
Testing Done
Tested on Local docker setup:
Ran the new Replication Job with Local Docker setup. Added a table with replicationConfig.
Observed:
For all the boxes checked, include a detailed description of the testing done for the changes made in this pull request.
Additional Information
For all the boxes checked, include additional details of the changes made in this pull request.