@Test public void testGetWriterFilePath() { Extract extract = new Extract(TableType.SNAPSHOT_ONLY, "org.apache.gobblin.dbNamespace", "tableName"); WorkUnit state = WorkUnit.create(extract); state.setProp(ConfigurationKeys.WRITER_FILE_PATH, TEST_WRITER_FILE_PATH); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 0, 0), TEST_WRITER_FILE_PATH); state.setProp(ConfigurationKeys.WRITER_FILE_PATH + ".0", TEST_WRITER_FILE_PATH); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 1, 1), TEST_WRITER_FILE_PATH); state.removeProp(ConfigurationKeys.WRITER_FILE_PATH); state.setProp(ConfigurationKeys.WRITER_FILE_PATH_TYPE, "tablename"); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 0, 0), new Path("tableName")); state.setProp(ConfigurationKeys.WRITER_FILE_PATH_TYPE, "namespace_table"); Assert.assertEquals(WriterUtils.getWriterFilePath(state, 0, 0), new Path("org/apache/gobblin/dbNamespace/tableName")); }
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; }