@Test public void testGetDefaultWriterFilePath() { String namespace = "gobblin.test"; String tableName = "test-table"; SourceState sourceState = new SourceState(); WorkUnit state = WorkUnit.create(new Extract(sourceState, TableType.APPEND_ONLY, namespace, tableName)); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 0, 0), new Path(state.getExtract().getOutputFilePath())); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 2, 0), new Path(state.getExtract().getOutputFilePath(), ConfigurationKeys.DEFAULT_FORK_BRANCH_NAME + "0")); }
public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) { if (state instanceof WorkUnitState) { WorkUnitState workUnitState = (WorkUnitState) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(), numBranches, branchId)); } else if (state instanceof WorkUnit) { WorkUnit workUnit = (WorkUnit) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(), numBranches, branchId)); } throw new RuntimeException("In order to get the default value for " + ConfigurationKeys.WRITER_FILE_PATH + " the given state must be of type " + WorkUnitState.class.getName() + " or " + WorkUnit.class.getName()); }
WriterUtils { public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) { if (state instanceof WorkUnitState) { WorkUnitState workUnitState = (WorkUnitState) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(), numBranches, branchId)); } else if (state instanceof WorkUnit) { WorkUnit workUnit = (WorkUnit) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(), numBranches, branchId)); } throw new RuntimeException("In order to get the default value for " + ConfigurationKeys.WRITER_FILE_PATH + " the given state must be of type " + WorkUnitState.class.getName() + " or " + WorkUnit.class.getName()); } }
WriterUtils { public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) { if (state instanceof WorkUnitState) { WorkUnitState workUnitState = (WorkUnitState) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(), numBranches, branchId)); } else if (state instanceof WorkUnit) { WorkUnit workUnit = (WorkUnit) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(), numBranches, branchId)); } throw new RuntimeException("In order to get the default value for " + ConfigurationKeys.WRITER_FILE_PATH + " the given state must be of type " + WorkUnitState.class.getName() + " or " + WorkUnit.class.getName()); }  }
WriterUtils { public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) { if (state instanceof WorkUnitState) { WorkUnitState workUnitState = (WorkUnitState) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(), numBranches, branchId)); } else if (state instanceof WorkUnit) { WorkUnit workUnit = (WorkUnit) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(), numBranches, branchId)); } throw new RuntimeException("In order to get the default value for " + ConfigurationKeys.WRITER_FILE_PATH + " the given state must be of type " + WorkUnitState.class.getName() + " or " + WorkUnit.class.getName()); }  static Path getWriterStagingDir(State state, int numBranches, int branchId); static Path getWriterStagingDir(State state, int numBranches, int branchId, String attemptId); static Path getWriterOutputDir(State state, int numBranches, int branchId); static Path getDataPublisherFinalDir(State state, int numBranches, int branchId); static Path getWriterFilePath(State state, int numBranches, int branchId); static Path getNamespaceTableWriterFilePath(State state); static Path getTableNameWriterFilePath(State state); static Path getDefaultWriterFilePath(State state, int numBranches, int branchId); static String getWriterFileName(State state, int numBranches, int branchId, String writerId,
      String formatExtension); static CodecFactory getCodecFactory(Optional<String> codecName, Optional<String> deflateLevel); static void mkdirsWithRecursivePermission(final FileSystem fs, final Path path, FsPermission perm); static void mkdirsWithRecursivePermissionWithRetry(final FileSystem fs, final Path path, FsPermission perm, Config retrierConfig); static URI getWriterFsUri(State state, int numBranches, int branchId); static FileSystem getWriterFS(State state, int numBranches, int branchId); static FileSystem getWriterFs(State state); static Configuration getFsConfiguration(State state); }
WriterUtils { public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) { if (state instanceof WorkUnitState) { WorkUnitState workUnitState = (WorkUnitState) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(), numBranches, branchId)); } else if (state instanceof WorkUnit) { WorkUnit workUnit = (WorkUnit) state; return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(), numBranches, branchId)); } throw new RuntimeException("In order to get the default value for " + ConfigurationKeys.WRITER_FILE_PATH + " the given state must be of type " + WorkUnitState.class.getName() + " or " + WorkUnit.class.getName()); }  static Path getWriterStagingDir(State state, int numBranches, int branchId); static Path getWriterStagingDir(State state, int numBranches, int branchId, String attemptId); static Path getWriterOutputDir(State state, int numBranches, int branchId); static Path getDataPublisherFinalDir(State state, int numBranches, int branchId); static Path getWriterFilePath(State state, int numBranches, int branchId); static Path getNamespaceTableWriterFilePath(State state); static Path getTableNameWriterFilePath(State state); static Path getDefaultWriterFilePath(State state, int numBranches, int branchId); static String getWriterFileName(State state, int numBranches, int branchId, String writerId,
      String formatExtension); static CodecFactory getCodecFactory(Optional<String> codecName, Optional<String> deflateLevel); static void mkdirsWithRecursivePermission(final FileSystem fs, final Path path, FsPermission perm); static void mkdirsWithRecursivePermissionWithRetry(final FileSystem fs, final Path path, FsPermission perm, Config retrierConfig); static URI getWriterFsUri(State state, int numBranches, int branchId); static FileSystem getWriterFS(State state, int numBranches, int branchId); static FileSystem getWriterFs(State state); static Configuration getFsConfiguration(State state); static final String WRITER_ENCRYPTED_CONFIG_PATH; static final Config NO_RETRY_CONFIG; }