Reorganize the code for hive registration to isolate function
Closes from ZihanLi58/ETL-8815
@@ -76,9 +76,9 @@ public class HiveRegistrationPublisher extends DataPublisher {
private static final boolean DEFAULT_PATH_DEDUPE_ENABLED = true;
private final Closer closer = Closer.create();
- private final HiveRegister hiveRegister;
- private final ExecutorService hivePolicyExecutor;
- private final MetricContext metricContext;
+ protected final HiveRegister hiveRegister;
+ protected final ExecutorService hivePolicyExecutor;
+ protected final MetricContext metricContext;
/**
* The configuration to determine if path deduplication should be enabled during Hive Registration process.
@@ -88,7 +88,7 @@ public class HiveRegistrationPublisher extends DataPublisher {
*
* e.g. In streaming mode, there could be cases that files(e.g. avro) under single topic folder carry different schema.
*/
- private boolean isPathDedupeEnabled;
+ protected boolean isPathDedupeEnabled;
/**
* Make the deduplication of path to be registered in the Publisher level,
@@ -118,18 +118,7 @@ public class HiveRegistrationPublisher extends DataPublisher {
}
}
- @Deprecated
- @Override
- public void initialize() throws IOException {}
-
- /**
- * @param states This is a collection of TaskState.
- */
- @Override
- public void publishData(Collection<? extends WorkUnitState> states) throws IOException {
- CompletionService<Collection<HiveSpec>> completionService =
- new ExecutorCompletionService<>(this.hivePolicyExecutor);
-
+ protected int computeSpecs(Collection<? extends WorkUnitState> states, CompletionService<Collection<HiveSpec>> completionService) {
// Each state in states is task-level State, while superState is the Job-level State.
// Using both State objects to distinguish each HiveRegistrationPolicy so that
// they can carry task-level information to pass into Hive Partition and its corresponding Hive Table.
@@ -154,10 +143,13 @@ public class HiveRegistrationPublisher extends DataPublisher {
final HiveRegistrationPolicy policy = HiveRegistrationPolicyBase.getPolicy(taskSpecificState);
for ( final String path : state.getPropAsList(ConfigurationKeys.PUBLISHER_DIRS) ) {
- if (isPathDedupeEnabled && pathsToRegisterFromSingleState.contains(path)){
+ if (isPathDedupeEnabled){
+ if (pathsToRegisterFromSingleState.contains(path)) {
continue;
- }
+ } else {
pathsToRegisterFromSingleState.add(path);
+ }
+ }
toRegisterPathCount += 1;
completionService.submit(new Callable<Collection<HiveSpec>>() {
@Override
@@ -169,8 +161,22 @@ public class HiveRegistrationPublisher extends DataPublisher {
});
}
}
- else continue;
}
+ return toRegisterPathCount;
+ }
+ @Deprecated
+ @Override
+ public void initialize() throws IOException {}
+
+ /**
+ * @param states This is a collection of TaskState.
+ */
+ @Override
+ public void publishData(Collection<? extends WorkUnitState> states) throws IOException {
+ CompletionService<Collection<HiveSpec>> completionService =
+ new ExecutorCompletionService<>(this.hivePolicyExecutor);
+
+ int toRegisterPathCount = computeSpecs(states, completionService);
for (int i = 0; i < toRegisterPathCount; i++) {
try {
for (HiveSpec spec : completionService.take().get()) {
