public static void persistTaskMetadata(HelixPropertyStore<ZNRecord> propertyStore, String taskType,
      BaseTaskMetadata taskMetadata, int expectedVersion) {
    String newPath =
        ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadata(taskMetadata.getTableNameWithType(),
            taskType);
    String oldPath = ZKMetadataProvider.constructPropertyStorePathForMinionTaskMetadataDeprecated(taskType,
        taskMetadata.getTableNameWithType());
    if (propertyStore.exists(newPath, AccessOption.PERSISTENT) || !propertyStore.exists(oldPath,
        AccessOption.PERSISTENT)) {
      persistTaskMetadata(newPath, propertyStore, taskType, taskMetadata, expectedVersion);
    } else {
      persistTaskMetadata(oldPath, propertyStore, taskType, taskMetadata, expectedVersion);
    }
  }
@Test
  public void testPersistTaskMetadataWithException() {
    DummyTaskMetadata taskMetadata = new DummyTaskMetadata(TABLE_NAME_WITH_TYPE, 1000);
    HelixPropertyStore<ZNRecord> mockPropertyStore = Mockito.mock(HelixPropertyStore.class);
    String expectedPath = NEW_MINION_METADATA_PATH;

    // Test happy path. No exceptions thrown.
    when(mockPropertyStore.set(expectedPath, taskMetadata.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION)).thenReturn(
        true);
    MinionTaskMetadataUtils.persistTaskMetadata(mockPropertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
    verify(mockPropertyStore, times(1)).set(expectedPath, taskMetadata.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);

    // Test exception thrown
    when(mockPropertyStore.set(expectedPath, taskMetadata.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION)).thenReturn(
        false);
    try {
      MinionTaskMetadataUtils.persistTaskMetadata(mockPropertyStore, TASK_TYPE, taskMetadata, EXPECTED_VERSION);
      fail("ZkException should have been thrown");
    } catch (ZkException e) {
      verify(mockPropertyStore, times(2)).set(expectedPath, taskMetadata.toZNRecord(), EXPECTED_VERSION, ACCESS_OPTION);
      assertEquals(e.getMessage(), "Failed to persist minion metadata for task: TestTaskType and metadata:"
          + " {\"tableNameWithType\":\"TestTable_OFFLINE\"}");
    }
  }