@Test public void testGetDefaultWriterFilePathWithWorkUnitState() { String namespace = "gobblin.test"; String tableName = "test-table"; SourceState sourceState = new SourceState(); WorkUnit workUnit = WorkUnit.create(new Extract(sourceState, TableType.APPEND_ONLY, namespace, tableName)); WorkUnitState workUnitState = new WorkUnitState(workUnit); Assert.assertEquals(WriterUtils.getWriterFilePath(workUnitState, 0, 0), new Path(workUnitState.getExtract() .getOutputFilePath())); Assert.assertEquals(WriterUtils.getWriterFilePath(workUnitState, 2, 0), new Path(workUnitState.getExtract() .getOutputFilePath(), ConfigurationKeys.DEFAULT_FORK_BRANCH_NAME + "0")); }
public static Path getWriterFilePath(State state, int numBranches, int branchId) { if (state.contains( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))); } switch (getWriterFilePathType(state)) { case NAMESPACE_TABLE: return getNamespaceTableWriterFilePath(state); case TABLENAME: return WriterUtils.getTableNameWriterFilePath(state); default: return WriterUtils.getDefaultWriterFilePath(state, numBranches, branchId); } }
WriterUtils { public static Path getWriterFilePath(State state, int numBranches, int branchId) { if (state.contains( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))); } switch (getWriterFilePathType(state)) { case NAMESPACE_TABLE: return getNamespaceTableWriterFilePath(state); case TABLENAME: return WriterUtils.getTableNameWriterFilePath(state); default: return WriterUtils.getDefaultWriterFilePath(state, numBranches, branchId); } } }
WriterUtils { public static Path getWriterFilePath(State state, int numBranches, int branchId) { if (state.contains( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))); } switch (getWriterFilePathType(state)) { case NAMESPACE_TABLE: return getNamespaceTableWriterFilePath(state); case TABLENAME: return WriterUtils.getTableNameWriterFilePath(state); default: return WriterUtils.getDefaultWriterFilePath(state, numBranches, branchId); } }  }
WriterUtils { public static Path getWriterFilePath(State state, int numBranches, int branchId) { if (state.contains( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))); } switch (getWriterFilePathType(state)) { case NAMESPACE_TABLE: return getNamespaceTableWriterFilePath(state); case TABLENAME: return WriterUtils.getTableNameWriterFilePath(state); default: return WriterUtils.getDefaultWriterFilePath(state, numBranches, branchId); } }  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 getWriterFilePath(State state, int numBranches, int branchId) { if (state.contains( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.WRITER_FILE_PATH, numBranches, branchId))); } switch (getWriterFilePathType(state)) { case NAMESPACE_TABLE: return getNamespaceTableWriterFilePath(state); case TABLENAME: return WriterUtils.getTableNameWriterFilePath(state); default: return WriterUtils.getDefaultWriterFilePath(state, numBranches, branchId); } }  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; }