@Override
  public void receivedEvent(Event event, Map<String, Object> metadata) {
    log.debug("received event {}, metadata {}", event, metadata);
    if (event instanceof ResourceEvent) {
      final ResourceAction action = ((ResourceEvent) event).getAction();
      final Optional<Class<? extends BaseResource<?, ?, ?, ?, ?>>> resource =
          getResourceClass(metadata);
      final Optional<String> namespaceOptional = event.getRelatedCustomResourceID().getNamespace();
      resource.ifPresent(
          aClass -> getCounter(aClass, action.name().toLowerCase(), RESOURCE, EVENT).inc());
      if (resource.isPresent() && namespaceOptional.isPresent()) {
        getCounter(
                resource.get(),
                namespaceOptional.get(),
                action.name().toLowerCase(),
                RESOURCE,
                EVENT)
            .inc();
      }
    }
  }
@Test
  void testReceivedEvent() {
    Event event = new ResourceEvent(ResourceAction.ADDED, resourceId, buildNamespacedResource());
    operatorMetrics.receivedEvent(event, metadata);
    Map<String, Metric> metrics = operatorMetrics.metricRegistry().getMetrics();
    assertEquals(2, metrics.size());
    assertTrue(metrics.containsKey("sparkapplication.added.resource.event"));
    assertTrue(metrics.containsKey("sparkapplication.testns.added.resource.event"));
  }