[apache#2821] feat(core): supports metalake event for event listener (a…

…pache#2897)

### What changes were proposed in this pull request?

supports metalake event for event listener

### Why are the changes needed?


Fix: apache#2821 

### Does this PR introduce _any_ user-facing change?
no

### How was this patch tested?
existing tests

core/src/main/java/com/datastrato/gravitino/GravitinoEnv.java

@@ -22,6 +22,8 @@
 import com.datastrato.gravitino.listener.EventBus;
 import com.datastrato.gravitino.listener.EventListenerManager;
 import com.datastrato.gravitino.lock.LockManager;
+import com.datastrato.gravitino.metalake.MetalakeDispatcher;
+import com.datastrato.gravitino.metalake.MetalakeEventDispatcher;
 import com.datastrato.gravitino.metalake.MetalakeManager;
 import com.datastrato.gravitino.metrics.MetricsSystem;
 import com.datastrato.gravitino.metrics.source.JVMMetricsSource;
@@ -55,7 +57,7 @@ public class GravitinoEnv {
 
   private TopicOperationDispatcher topicOperationDispatcher;
 
-  private MetalakeManager metalakeManager;
+  private MetalakeDispatcher metalakeDispatcher;
 
   private AccessControlManager accessControlManager;
 
@@ -131,7 +133,8 @@ public void initialize(Config config) {
     EventBus eventBus = eventListenerManager.createEventBus();
 
     // Create and initialize metalake related modules
-    this.metalakeManager = new MetalakeManager(entityStore, idGenerator);
+    MetalakeManager metalakeManager = new MetalakeManager(entityStore, idGenerator);
+    this.metalakeDispatcher = new MetalakeEventDispatcher(eventBus, metalakeManager);
 
     // Create and initialize Catalog related modules
     this.catalogManager = new CatalogManager(config, entityStore, idGenerator);
@@ -231,12 +234,12 @@ public TopicOperationDispatcher topicOperationDispatcher() {
   }
 
   /**
-   * Get the MetalakeManager associated with the Gravitino environment.
+   * Get the MetalakeDispatcher associated with the Gravitino environment.
    *
-   * @return The MetalakeManager instance.
+   * @return The MetalakeDispatcher instance.
    */
-  public MetalakeManager metalakesManager() {
-    return metalakeManager;
+  public MetalakeDispatcher metalakeDispatcher() {
+    return metalakeDispatcher;
   }
 
   /**

core/src/main/java/com/datastrato/gravitino/listener/api/event/AlterMetalakeEvent.java

@@ -0,0 +1,58 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.MetalakeChange;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+import com.datastrato.gravitino.listener.api.info.MetalakeInfo;
+
+/** Represents an event fired when a metalake is successfully altered. */
+@DeveloperApi
+public final class AlterMetalakeEvent extends MetalakeEvent {
+  private final MetalakeInfo updatedMetalakeInfo;
+  private final MetalakeChange[] metalakeChanges;
+
+  /**
+   * Constructs an instance of {@code AlterMetalakeEvent}, encapsulating the key details about the
+   * successful alteration of a metalake.
+   *
+   * @param user The username of the individual responsible for initiating the metalake alteration.
+   * @param identifier The unique identifier of the altered metalake, serving as a clear reference
+   *     point for the metalake in question.
+   * @param metalakeChanges An array of {@link MetalakeChange} objects representing the specific
+   *     changes applied to the metalake during the alteration process.
+   * @param updatedMetalakeInfo The post-alteration state of the metalake.
+   */
+  public AlterMetalakeEvent(
+      String user,
+      NameIdentifier identifier,
+      MetalakeChange[] metalakeChanges,
+      MetalakeInfo updatedMetalakeInfo) {
+    super(user, identifier);
+    this.metalakeChanges = metalakeChanges.clone();
+    this.updatedMetalakeInfo = updatedMetalakeInfo;
+  }
+
+  /**
+   * Retrieves the updated state of the metalake after the successful alteration.
+   *
+   * @return A {@link MetalakeInfo} instance encapsulating the details of the altered metalake.
+   */
+  public MetalakeInfo updatedMetalakeInfo() {
+    return updatedMetalakeInfo;
+  }
+
+  /**
+   * Retrieves the specific changes that were made to the metalake during the alteration process.
+   *
+   * @return An array of {@link MetalakeChange} objects detailing each modification applied to the
+   *     metalake.
+   */
+  public MetalakeChange[] metalakeChanges() {
+    return metalakeChanges;
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/AlterMetalakeFailureEvent.java

@@ -0,0 +1,47 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.MetalakeChange;
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+
+/**
+ * Represents an event that is triggered when an attempt to alter a metalake fails due to an
+ * exception.
+ */
+@DeveloperApi
+public final class AlterMetalakeFailureEvent extends MetalakeFailureEvent {
+  private final MetalakeChange[] metalakeChanges;
+
+  /**
+   * Constructs an {@code AlterMetalakeFailureEvent} instance, capturing detailed information about
+   * the failed metalake alteration attempt.
+   *
+   * @param user The user who initiated the metalake alteration operation.
+   * @param identifier The identifier of the metalake that was attempted to be altered.
+   * @param exception The exception that was thrown during the metalake alteration operation.
+   * @param metalakeChanges The changes that were attempted on the metalake.
+   */
+  public AlterMetalakeFailureEvent(
+      String user,
+      NameIdentifier identifier,
+      Exception exception,
+      MetalakeChange[] metalakeChanges) {
+    super(user, identifier, exception);
+    this.metalakeChanges = metalakeChanges.clone();
+  }
+
+  /**
+   * Retrieves the changes that were attempted on the metalake.
+   *
+   * @return An array of {@link MetalakeChange} objects representing the attempted modifications to
+   *     the metalake.
+   */
+  public MetalakeChange[] metalakeChanges() {
+    return metalakeChanges;
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/CreateMetalakeEvent.java

@@ -0,0 +1,40 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+import com.datastrato.gravitino.listener.api.info.MetalakeInfo;
+
+/** Represents an event triggered upon the successful creation of a Metalake. */
+@DeveloperApi
+public final class CreateMetalakeEvent extends MetalakeEvent {
+  private final MetalakeInfo createdMetalakeInfo;
+  /**
+   * Constructs an instance of {@code CreateMetalakeEvent}, capturing essential details about the
+   * successful creation of a metalake.
+   *
+   * @param user The username of the individual who initiated the metalake creation.
+   * @param identifier The unique identifier of the metalake that was created.
+   * @param createdMetalakeInfo The final state of the metalake post-creation.
+   */
+  public CreateMetalakeEvent(
+      String user, NameIdentifier identifier, MetalakeInfo createdMetalakeInfo) {
+    super(user, identifier);
+    this.createdMetalakeInfo = createdMetalakeInfo;
+  }
+
+  /**
+   * Retrieves the final state of the Metalake as it was returned to the user after successful
+   * creation.
+   *
+   * @return A {@link MetalakeInfo} instance encapsulating the comprehensive details of the newly
+   *     created Metalake.
+   */
+  public MetalakeInfo createdMetalakeInfo() {
+    return createdMetalakeInfo;
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/CreateMetalakeFailureEvent.java

@@ -0,0 +1,38 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+import com.datastrato.gravitino.listener.api.info.MetalakeInfo;
+
+/**
+ * Represents an event that is generated when an attempt to create a Metalake fails due to an
+ * exception.
+ */
+@DeveloperApi
+public final class CreateMetalakeFailureEvent extends MetalakeFailureEvent {
+  private final MetalakeInfo createMetalakeRequest;
+
+  public CreateMetalakeFailureEvent(
+      String user,
+      NameIdentifier identifier,
+      Exception exception,
+      MetalakeInfo createMetalakeRequest) {
+    super(user, identifier, exception);
+    this.createMetalakeRequest = createMetalakeRequest;
+  }
+
+  /**
+   * Retrieves the original request information for the attempted Metalake creation.
+   *
+   * @return The {@link MetalakeInfo} instance representing the request information for the failed
+   *     Metalake creation attempt.
+   */
+  public MetalakeInfo createMetalakeRequest() {
+    return createMetalakeRequest;
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/DropMetalakeEvent.java

@@ -0,0 +1,41 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+
+/**
+ * Represents an event that is generated after a Metalake is successfully removed from the system.
+ */
+@DeveloperApi
+public final class DropMetalakeEvent extends MetalakeEvent {
+  private final boolean isExists;
+
+  /**
+   * Constructs a new {@code DropMetalakeEvent} instance, encapsulating information about the
+   * outcome of a metalake drop operation.
+   *
+   * @param user The user who initiated the drop metalake operation.
+   * @param identifier The identifier of the metalake that was attempted to be dropped.
+   * @param isExists A boolean flag indicating whether the metalake existed at the time of the drop
+   *     operation.
+   */
+  public DropMetalakeEvent(String user, NameIdentifier identifier, boolean isExists) {
+    super(user, identifier);
+    this.isExists = isExists;
+  }
+
+  /**
+   * Retrieves the existence status of the Metalake at the time of the removal operation.
+   *
+   * @return A boolean value indicating whether the Metalake existed. {@code true} if the Metalake
+   *     existed, otherwise {@code false}.
+   */
+  public boolean isExists() {
+    return isExists;
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/DropMetalakeFailureEvent.java

@@ -0,0 +1,29 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+
+/**
+ * Represents an event that is generated when an attempt to remove a Metalake from the system fails
+ * due to an exception.
+ */
+@DeveloperApi
+public final class DropMetalakeFailureEvent extends MetalakeFailureEvent {
+  /**
+   * Constructs a new {@code DropMetalakeFailureEvent} instance, capturing detailed information
+   * about the failed attempt to drop a metalake.
+   *
+   * @param user The user who initiated the drop metalake operation.
+   * @param identifier The identifier of the metalake that the operation attempted to drop.
+   * @param exception The exception that was thrown during the drop metalake operation, offering
+   *     insights into what went wrong and why the operation failed.
+   */
+  public DropMetalakeFailureEvent(String user, NameIdentifier identifier, Exception exception) {
+    super(user, identifier, exception);
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/ListMetalakeEvent.java

@@ -0,0 +1,21 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.annotation.DeveloperApi;
+
+/** Represents an event that is triggered upon the successful list of metalakes. */
+@DeveloperApi
+public final class ListMetalakeEvent extends MetalakeEvent {
+  /**
+   * Constructs an instance of {@code ListMetalakeEvent}.
+   *
+   * @param user The username of the individual who initiated the metalake listing.
+   */
+  public ListMetalakeEvent(String user) {
+    super(user, null);
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/ListMetalakeFailureEvent.java

@@ -0,0 +1,26 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.annotation.DeveloperApi;
+
+/**
+ * Represents an event that is triggered when an attempt to list metalakes fails due to an
+ * exception.
+ */
+@DeveloperApi
+public final class ListMetalakeFailureEvent extends MetalakeFailureEvent {
+
+  /**
+   * Constructs a {@code ListMetalakeFailureEvent} instance.
+   *
+   * @param user The username of the individual who initiated the operation to list metalakes.
+   * @param exception The exception encountered during the attempt to list metalakes.
+   */
+  public ListMetalakeFailureEvent(String user, Exception exception) {
+    super(user, null, exception);
+  }
+}

core/src/main/java/com/datastrato/gravitino/listener/api/event/LoadMetalakeEvent.java

@@ -0,0 +1,38 @@
+/*
+ *  Copyright 2024 Datastrato Pvt Ltd.
+ *  This software is licensed under the Apache License version 2.
+ */
+
+package com.datastrato.gravitino.listener.api.event;
+
+import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.annotation.DeveloperApi;
+import com.datastrato.gravitino.listener.api.info.MetalakeInfo;
+
+/** Represents an event that is generated when a Metalake is successfully loaded. */
+@DeveloperApi
+public final class LoadMetalakeEvent extends MetalakeEvent {
+  private final MetalakeInfo loadedMetalakeInfo;
+
+  /**
+   * Constructs an instance of {@code LoadMetalakeEvent}.
+   *
+   * @param user The username of the individual who initiated the metalake loading.
+   * @param identifier The unique identifier of the metalake that was loaded.
+   * @param metalakeInfo The state of the metalake post-loading.
+   */
+  public LoadMetalakeEvent(String user, NameIdentifier identifier, MetalakeInfo metalakeInfo) {
+    super(user, identifier);
+    this.loadedMetalakeInfo = metalakeInfo;
+  }
+
+  /**
+   * Retrieves detailed information about the Metalake that was successfully loaded.
+   *
+   * @return A {@link MetalakeInfo} instance containing comprehensive details of the Metalake,
+   *     including its configuration, properties, and state at the time of loading.
+   */
+  public MetalakeInfo loadedMetalakeInfo() {
+    return loadedMetalakeInfo;
+  }
+}