fix(broker): report health on change only

Previously every invocation of updateHealthStatus caused listeners to be called even if nothing changed.

(cherry picked from commit e5cf4d8)

broker/src/main/java/io/camunda/zeebe/broker/system/partitions/ZeebePartitionHealth.java

@@ -12,6 +12,7 @@
 import io.camunda.zeebe.util.health.HealthReport;
 import io.camunda.zeebe.util.health.HealthStatus;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -57,19 +58,12 @@ private void updateHealthStatus() {
       healthReport = HealthReport.healthy(this);
     }
 
-    if (previousStatus != healthReport) {
+    if (!Objects.equals(previousStatus, healthReport)) {
       switch (healthReport.getStatus()) {
-        case HEALTHY:
-          failureListeners.forEach(FailureListener::onRecovered);
-          break;
-        case UNHEALTHY:
-          failureListeners.forEach((l) -> l.onFailure(healthReport));
-          break;
-        case DEAD:
-          failureListeners.forEach((l) -> l.onUnrecoverableFailure(healthReport));
-          break;
-        default:
-          break;
+        case HEALTHY -> failureListeners.forEach(FailureListener::onRecovered);
+        case UNHEALTHY -> failureListeners.forEach((l) -> l.onFailure(healthReport));
+        case DEAD -> failureListeners.forEach((l) -> l.onUnrecoverableFailure(healthReport));
+        default -> {}
       }
     }
   }

broker/src/test/java/io/camunda/zeebe/broker/system/partitions/ZeebePartitionTest.java

@@ -33,6 +33,7 @@
 import io.camunda.zeebe.scheduler.testing.ControlledActorSchedulerRule;
 import io.camunda.zeebe.util.exception.UnrecoverableException;
 import io.camunda.zeebe.util.health.FailureListener;
+import io.camunda.zeebe.util.health.HealthIssue;
 import io.camunda.zeebe.util.health.HealthReport;
 import io.camunda.zeebe.util.health.HealthStatus;
 import java.time.Duration;
@@ -400,6 +401,91 @@ public void shouldReportUnhealthyPerDefault() {
     assertThat(healthReport.getIssue().message()).contains("Services not installed");
   }
 
+  @Test
+  public void shouldCallOnFailureOnceForSameHealthIssue() {
+    // given
+    schedulerRule.submitActor(partition);
+    schedulerRule.workUntilDone();
+
+    final FailureListener failureListener = mock(FailureListener.class);
+    doNothing().when(failureListener).onFailure(any());
+
+    final var captor = ArgumentCaptor.forClass(ZeebePartitionHealth.class);
+    verify(healthMonitor).registerComponent(any(), captor.capture());
+    final var zeebePartitionHealth = captor.getValue();
+    zeebePartitionHealth.addFailureListener(failureListener);
+
+    when(transition.getHealthIssue()).thenReturn(HealthIssue.of("it's over"));
+
+    // when
+    final HealthReport healthReport1 = zeebePartitionHealth.getHealthReport();
+    assertThat(healthReport1.getStatus()).isEqualTo(HealthStatus.UNHEALTHY);
+
+    final HealthReport healthReport2 = zeebePartitionHealth.getHealthReport();
+
+    // then
+    assertThat(healthReport1).isEqualTo(healthReport2);
+    verify(failureListener, times(1)).onFailure(any());
+  }
+
+  @Test
+  public void shouldCallOnFailureOnHealthIssueChange() {
+    // given
+    schedulerRule.submitActor(partition);
+    schedulerRule.workUntilDone();
+
+    final FailureListener failureListener = mock(FailureListener.class);
+    doNothing().when(failureListener).onFailure(any());
+
+    final var captor = ArgumentCaptor.forClass(ZeebePartitionHealth.class);
+    verify(healthMonitor).registerComponent(any(), captor.capture());
+    final var zeebePartitionHealth = captor.getValue();
+    zeebePartitionHealth.addFailureListener(failureListener);
+
+    when(transition.getHealthIssue()).thenReturn(HealthIssue.of("it's over"));
+
+    // when
+    final HealthReport healthReport1 = zeebePartitionHealth.getHealthReport();
+    assertThat(healthReport1.getStatus()).isEqualTo(HealthStatus.UNHEALTHY);
+
+    when(transition.getHealthIssue()).thenReturn(HealthIssue.of("it's something else"));
+    final HealthReport healthReport2 = zeebePartitionHealth.getHealthReport();
+    assertThat(healthReport2.getStatus()).isEqualTo(HealthStatus.UNHEALTHY);
+
+    // then
+    assertThat(healthReport1).isNotEqualTo(healthReport2);
+    verify(failureListener, times(2)).onFailure(any());
+  }
+
+  @Test
+  public void shouldCallOnRecoveredOnceWhenHealthy() {
+    // given
+    schedulerRule.submitActor(partition);
+    schedulerRule.workUntilDone();
+
+    final FailureListener failureListener = mock(FailureListener.class);
+    doNothing().when(failureListener).onFailure(any());
+    doNothing().when(failureListener).onRecovered();
+
+    final var captor = ArgumentCaptor.forClass(ZeebePartitionHealth.class);
+    verify(healthMonitor).registerComponent(any(), captor.capture());
+    final var zeebePartitionHealth = captor.getValue();
+    zeebePartitionHealth.addFailureListener(failureListener);
+
+    // when
+    partition.onNewRole(Role.LEADER, 1);
+    schedulerRule.workUntilDone();
+
+    final HealthReport healthReport1 = zeebePartitionHealth.getHealthReport();
+    assertThat(healthReport1.getStatus()).isEqualTo(HealthStatus.HEALTHY);
+
+    final HealthReport healthReport2 = zeebePartitionHealth.getHealthReport();
+    assertThat(healthReport2.getStatus()).isEqualTo(HealthStatus.HEALTHY);
+    // then
+    assertThat(healthReport1).isEqualTo(healthReport2);
+    verify(failureListener, times(1)).onRecovered();
+  }
+
   @Test
   public void shouldReportHealthyAfterTransition() {
     // given