public boolean handleSentinelResourceReconciliation(CR resource, KubernetesClient client) {
    if (!isSentinelResource(resource)) {
      return false;
    }

    ResourceID resourceId = ResourceID.fromResource(resource);
    sentinelResources.compute(
        resourceId,
        (id, previousState) -> {
          boolean firstReconcile = false;
          if (previousState == null) {
            firstReconcile = true;
            previousState = new SentinelResourceState();
          }
          previousState.onReconcile(resource);
          if (firstReconcile) {
            updateSpecAndScheduleHealthCheck(resourceId, previousState, client);
          }
          return previousState;
        });
    return true;
  }
@Test
  @Order(3)
  void testHandleSentinelResourceReconciliation() throws InterruptedException {
    // Reduce the SENTINEL_RESOURCE_RECONCILIATION_DELAY time to 0
    SparkOperatorConfManager.INSTANCE.refresh(
        Collections.singletonMap(
            SparkOperatorConf.SENTINEL_RESOURCE_RECONCILIATION_DELAY.getKey(), "10"));

    // Before Spark Reconciler Started
    SparkApplication mockApp = createMockApp(DEFAULT);
    kubernetesClient.resource(ReconcilerUtils.clone(mockApp)).create();
    KubernetesResourceList<SparkApplication> crList =
        kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list();
    SparkApplication sparkApplication = crList.getItems().get(0);
    Long generation = sparkApplication.getMetadata().getGeneration();
    assertEquals(generation, 1L);

    // Spark Reconciler Handle Sentinel Resources at the first time
    var sentinelManager = new SentinelManager<SparkApplication>();
    sentinelManager.handleSentinelResourceReconciliation(sparkApplication, kubernetesClient);
    KubernetesResourceList<SparkApplication> crList2 =
        kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list();
    SparkApplication sparkApplication2 = crList2.getItems().get(0);
    Map<String, String> sparkConf2 = new HashMap<>(sparkApplication2.getSpec().getSparkConf());
    long generation2 = sparkApplication2.getMetadata().getGeneration();

    assertEquals(sparkConf2.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD), "1");
    assertEquals(generation2, 2L);
    var state2 =
        (SentinelManager<SparkApplication>.SentinelResourceState)
            sentinelManager.getSentinelResources().get(ResourceID.fromResource(mockApp));
    long previousGeneration2 = state2.previousGeneration;
    assertTrue(sentinelManager.allSentinelsAreHealthy());
    assertEquals(previousGeneration2, 1L);

    Thread.sleep(Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS * 2).toMillis());
    List<SparkApplication> crList3 =
        kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).list().getItems();
    SparkApplication sparkApplication3 = crList3.get(0);
    Map<String, String> sparkConf3 = new HashMap<>(sparkApplication3.getSpec().getSparkConf());
    // Spark Sentinel Applications' s k8s generation should change
    assertNotEquals(sparkApplication3.getMetadata().getGeneration(), generation2);
    // Spark conf SPARK_CONF_SENTINEL_DUMMY_FIELD values should increase
    assertNotEquals(
        sparkConf2.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD),
        sparkConf3.get(Constants.SENTINEL_RESOURCE_DUMMY_FIELD));
    var state3 =
        (SentinelManager<SparkApplication>.SentinelResourceState)
            sentinelManager.getSentinelResources().get(ResourceID.fromResource(mockApp));
    assertEquals(state3.previousGeneration, previousGeneration2);
    // Given the 2 * SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS, the reconcile method is
    // not called to handleSentinelResourceReconciliation to update
    assertFalse(sentinelManager.allSentinelsAreHealthy());

    sentinelManager.handleSentinelResourceReconciliation(sparkApplication2, kubernetesClient);
    sentinelManager.handleSentinelResourceReconciliation(sparkApplication2, kubernetesClient);
    boolean isHealthy;
    long currentTimeInMills = System.currentTimeMillis();
    do {
      isHealthy = sentinelManager.allSentinelsAreHealthy();
    } while (!isHealthy
        && notTimedOut(
            currentTimeInMills,
            TimeUnit.MILLISECONDS.convert(
                Duration.ofSeconds(SENTINEL_RESOURCE_RECONCILIATION_DELAY_SECONDS))));
    assertTrue(isHealthy);
    kubernetesClient.resources(SparkApplication.class).inNamespace(DEFAULT).delete();
  }