public static String generateSparkAppId(final SparkApplication app) {
    long attemptId = ModelUtils.getAttemptId(app);
    String preferredId = String.format("%s-%d", app.getMetadata().getName(), attemptId);
    if (preferredId.length() > DEFAULT_ID_LENGTH_LIMIT) {
      int preferredIdPrefixLength =
          DEFAULT_ID_LENGTH_LIMIT - DEFAULT_HASH_BASED_IDENTIFIER_LENGTH_LIMIT - 1;
      String preferredIdPrefix = preferredId.substring(0, preferredIdPrefixLength);
      return generateHashBasedId(
          preferredIdPrefix,
          app.getMetadata().getNamespace(),
          app.getMetadata().getName(),
          String.valueOf(attemptId));
    } else {
      return preferredId;
    }
  }
@Test
  void sparkAppIdShouldBeDeterministicPerAppPerAttempt() {
    SparkApplication mockApp1 = mock(SparkApplication.class);
    SparkApplication mockApp2 = mock(SparkApplication.class);
    ApplicationStatus mockStatus1 = mock(ApplicationStatus.class);
    ApplicationStatus mockStatus2 = mock(ApplicationStatus.class);
    String appName1 = "app1";
    String appName2 = "app2";
    ObjectMeta appMeta1 = new ObjectMetaBuilder().withName(appName1).withNamespace("ns").build();
    ObjectMeta appMeta2 = new ObjectMetaBuilder().withName(appName2).withNamespace("ns").build();
    when(mockApp1.getMetadata()).thenReturn(appMeta1);
    when(mockApp2.getMetadata()).thenReturn(appMeta2);
    when(mockApp1.getStatus()).thenReturn(mockStatus1);
    when(mockApp2.getStatus()).thenReturn(mockStatus2);

    String appId1 = SparkAppSubmissionWorker.generateSparkAppId(mockApp1);
    String appId2 = SparkAppSubmissionWorker.generateSparkAppId(mockApp2);

    assertNotEquals(appId1, appId2);
    assertTrue(appId1.contains(appName1));
    assertTrue(appId1.length() <= DEFAULT_ID_LENGTH_LIMIT);
    assertTrue(appId2.length() <= DEFAULT_ID_LENGTH_LIMIT);
    // multiple invoke shall give same result
    assertEquals(
        appId1,
        SparkAppSubmissionWorker.generateSparkAppId(mockApp1),
        "Multiple invoke of generateSparkAppId shall give same result.");
    assertEquals(
        appId2,
        SparkAppSubmissionWorker.generateSparkAppId(mockApp2),
        "Multiple invoke of generateSparkAppId shall give same result.");

    ApplicationAttemptSummary mockAttempt = mock(ApplicationAttemptSummary.class);
    AttemptInfo mockAttemptInfo = mock(AttemptInfo.class);
    when(mockAttempt.getAttemptInfo()).thenReturn(mockAttemptInfo);
    when(mockAttemptInfo.getId()).thenReturn(2L);
    when(mockStatus1.getCurrentAttemptSummary()).thenReturn(mockAttempt);
    when(mockStatus2.getCurrentAttemptSummary()).thenReturn(mockAttempt);

    String appId1Attempt2 = SparkAppSubmissionWorker.generateSparkAppId(mockApp1);
    assertTrue(appId1Attempt2.contains(appName1));
    assertNotEquals(appId1, appId1Attempt2);
    assertTrue(appId1Attempt2.length() <= DEFAULT_ID_LENGTH_LIMIT);

    String appId2Attempt2 = SparkAppSubmissionWorker.generateSparkAppId(mockApp2);
    assertNotEquals(appId2, appId2Attempt2);
    assertEquals(appId2Attempt2, SparkAppSubmissionWorker.generateSparkAppId(mockApp2));
    assertTrue(appId2Attempt2.length() <= DEFAULT_ID_LENGTH_LIMIT);

    assertEquals(appId1Attempt2, SparkAppSubmissionWorker.generateSparkAppId(mockApp1));
  }