[Improvement] CreateTableFailureEvent listener exception masks original createTable failure

What would you like to be improved?

In TableEventDispatcher#createTable, when dispatcher.createTable(...) throws, the code enters the catch block and dispatches CreateTableFailureEvent. If eventBus.dispatchEvent(...) also throws, that new exception replaces the original create-table exception. This can hide the real root cause from callers and make production failures much harder to diagnose.

Relevant path: TableEventDispatcher.java (line 141)

How should we improve?

Protect failure-event dispatch so it cannot mask the original exception. For example, wrap eventBus.dispatchEvent(new CreateTableFailureEvent(...)) in its own try/catch, preserve the original exception, and then rethrow the original failure.

Here's a unit test to help:

  @Test
  void testCreateTableFailureEventShouldNotMaskOriginalExceptionWhenListenerThrows() {
    NameIdentifier identifier = NameIdentifier.of("metalake", "catalog", table.name());
    GravitinoRuntimeException originalException =
        new GravitinoRuntimeException("Original create table failure");
    GravitinoRuntimeException listenerException =
        new GravitinoRuntimeException("Failure listener exception");
    TableDispatcher tableExceptionDispatcher = mock(TableDispatcher.class);
    when(tableExceptionDispatcher.createTable(
            any(NameIdentifier.class),
            any(Column[].class),
            any(String.class),
            any(Map.class),
            any(Transform[].class),
            any(Distribution.class),
            any(SortOrder[].class),
            any(Index[].class)))
        .thenThrow(originalException);

    EventBus eventBus =
        new EventBus(
            Arrays.asList(
                new EventListenerPlugin() {
                  @Override
                  public void init(Map<String, String> properties) {}

                  @Override
                  public void start() {}

                  @Override
                  public void stop() {}

                  @Override
                  public void onPostEvent(Event event) {
                    if (event instanceof CreateTableFailureEvent) {
                      throw listenerException;
                    }
                  }
                }));
    TableEventDispatcher tableEventDispatcher =
        new TableEventDispatcher(eventBus, tableExceptionDispatcher);

    GravitinoRuntimeException thrownException =
        Assertions.assertThrowsExactly(
            GravitinoRuntimeException.class,
            () ->
                tableEventDispatcher.createTable(
                    identifier,
                    table.columns(),
                    table.comment(),
                    table.properties(),
                    table.partitioning(),
                    table.distribution(),
                    table.sortOrder(),
                    table.index()));

    Assertions.assertSame(originalException, thrownException);
  }