@Test public void testGetDataPublisherFinalOutputDir() { State state = new State(); state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, TEST_DATA_PUBLISHER_FINAL_DIR); state.setProp(ConfigurationKeys.WRITER_FILE_PATH, TEST_WRITER_FILE_PATH); Assert.assertEquals(WriterUtils.getDataPublisherFinalDir(state, 0, 0), new Path(TEST_DATA_PUBLISHER_FINAL_DIR, TEST_WRITER_FILE_PATH)); state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR + ".0", TEST_DATA_PUBLISHER_FINAL_DIR); state.setProp(ConfigurationKeys.WRITER_FILE_PATH + ".0", TEST_WRITER_FILE_PATH); Assert.assertEquals(WriterUtils.getDataPublisherFinalDir(state, 2, 0), new Path(TEST_DATA_PUBLISHER_FINAL_DIR, TEST_WRITER_FILE_PATH)); state.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR + ".1", TEST_DATA_PUBLISHER_FINAL_DIR); state.setProp(ConfigurationKeys.WRITER_FILE_PATH + ".1", TEST_WRITER_FILE_PATH); Assert.assertEquals(WriterUtils.getDataPublisherFinalDir(state, 2, 1), new Path(TEST_DATA_PUBLISHER_FINAL_DIR, TEST_WRITER_FILE_PATH)); }
public static Path getDataPublisherFinalDir(State state, int numBranches, int branchId) { String dataPublisherFinalDirKey = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId); Preconditions.checkArgument(state.contains(dataPublisherFinalDirKey), "Missing required property " + dataPublisherFinalDirKey); if (state.getPropAsBoolean(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, ConfigurationKeys.DEFAULT_DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR)) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId)), WriterUtils.getWriterFilePath(state, numBranches, branchId)); } else { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId))); } }
WriterUtils { public static Path getDataPublisherFinalDir(State state, int numBranches, int branchId) { String dataPublisherFinalDirKey = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId); Preconditions.checkArgument(state.contains(dataPublisherFinalDirKey), "Missing required property " + dataPublisherFinalDirKey); if (state.getPropAsBoolean(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, ConfigurationKeys.DEFAULT_DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR)) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId)), WriterUtils.getWriterFilePath(state, numBranches, branchId)); } else { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId))); } } }
WriterUtils { public static Path getDataPublisherFinalDir(State state, int numBranches, int branchId) { String dataPublisherFinalDirKey = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId); Preconditions.checkArgument(state.contains(dataPublisherFinalDirKey), "Missing required property " + dataPublisherFinalDirKey); if (state.getPropAsBoolean(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, ConfigurationKeys.DEFAULT_DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR)) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId)), WriterUtils.getWriterFilePath(state, numBranches, branchId)); } else { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId))); } }  }
WriterUtils { public static Path getDataPublisherFinalDir(State state, int numBranches, int branchId) { String dataPublisherFinalDirKey = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId); Preconditions.checkArgument(state.contains(dataPublisherFinalDirKey), "Missing required property " + dataPublisherFinalDirKey); if (state.getPropAsBoolean(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, ConfigurationKeys.DEFAULT_DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR)) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId)), WriterUtils.getWriterFilePath(state, numBranches, branchId)); } else { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, 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 getDataPublisherFinalDir(State state, int numBranches, int branchId) { String dataPublisherFinalDirKey = ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId); Preconditions.checkArgument(state.contains(dataPublisherFinalDirKey), "Missing required property " + dataPublisherFinalDirKey); if (state.getPropAsBoolean(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, ConfigurationKeys.DEFAULT_DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR)) { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, numBranches, branchId)), WriterUtils.getWriterFilePath(state, numBranches, branchId)); } else { return new Path(state.getProp( ForkOperatorUtils.getPropertyNameForBranch(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, 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; }