fix

Author: zhangshenghang

docs/en/seatunnel-engine/rest-api-v1.md

@@ -843,46 +843,4 @@ Returns a list of logs from the requested node.
 To get a list of logs from the current node: `http://localhost:5801/hazelcast/rest/maps/log`
 To get the content of a log file: `http://localhost:5801/hazelcast/rest/maps/log/job-898380162133917698.log`
 
-</details>
-
-------------------------------------------------------------------------------------------
-
-### Return Job Event Information
-
-<details>
- <summary><code>GET</code> <code><b>/hazelcast/rest/maps/event/:jobId</b></code> <code>(Returns the list of job events)</code></summary>
-
-#### Parameters
-
-> | Parameter Name | Required | Type    | Description       |
-> |----------------|----------|---------|-------------------|
-> | jobId          | Yes      | string  | The Job ID        |
-
-#### Response
-
-```json
-[
-  {
-    "createdTime": 1739501227166,
-    "eventType": "LIFECYCLE_READER_OPEN"
-  },
-  {
-    "createdTime": 1739501227232,
-    "eventType": "LIFECYCLE_ENUMERATOR_OPEN"
-  },
-  {
-    "createdTime": 1739501227457,
-    "eventType": "LIFECYCLE_ENUMERATOR_CLOSE"
-  },
-  {
-    "createdTime": 1739501227516,
-    "eventType": "LIFECYCLE_WRITER_CLOSE"
-  },
-  {
-    "createdTime": 1739501228305,
-    "eventType": "LIFECYCLE_READER_CLOSE"
-  }
-]
-```
-
 </details>

docs/zh/seatunnel-engine/rest-api-v1.md

@@ -850,46 +850,3 @@ network:
 获取日志文件内容：`http://localhost:5801/hazelcast/rest/maps/log/job-898380162133917698.log`
 
 </details>
-
-
-------------------------------------------------------------------------------------------
-
-### 返回作业的事件信息
-
-<details>
- <summary><code>GET</code> <code><b>/hazelcast/rest/maps/event/:jobId</b></code> <code>(返回作业事件列表)</code></summary>
-
-#### 参数
-
-> | 参数名称  | 是否必传 |  参数类型  | 参数描述 |
-> |-------|------|--------|------|
-> | jobId | 是    | string | Job ID  |
-
-#### 响应
-
-```json
-[
-  {
-    "createdTime": 1739501227166,
-    "eventType": "LIFECYCLE_READER_OPEN"
-  },
-  {
-    "createdTime": 1739501227232,
-    "eventType": "LIFECYCLE_ENUMERATOR_OPEN"
-  },
-  {
-    "createdTime": 1739501227457,
-    "eventType": "LIFECYCLE_ENUMERATOR_CLOSE"
-  },
-  {
-    "createdTime": 1739501227516,
-    "eventType": "LIFECYCLE_WRITER_CLOSE"
-  },
-  {
-    "createdTime": 1739501228305,
-    "eventType": "LIFECYCLE_READER_CLOSE"
-  }
-]
-```
-
-</details>

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java

@@ -21,7 +21,6 @@
 
 import org.apache.seatunnel.api.common.metrics.JobMetrics;
 import org.apache.seatunnel.api.common.metrics.RawJobMetrics;
-import org.apache.seatunnel.api.event.Event;
 import org.apache.seatunnel.api.event.EventHandler;
 import org.apache.seatunnel.api.event.EventProcessor;
 import org.apache.seatunnel.api.tracing.MDCExecutorService;
@@ -80,7 +79,6 @@
 import com.hazelcast.spi.impl.NodeEngineImpl;
 import lombok.Getter;
 import lombok.NonNull;
-import lombok.Setter;
 import scala.Tuple2;
 
 import java.util.ArrayList;
@@ -92,7 +90,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -156,8 +153,6 @@ public class CoordinatorService {
      */
     @Getter private final Map<Long, JobMaster> runningJobMasterMap = new ConcurrentHashMap<>();
 
-    @Getter @Setter
-    private Map<Long, ArrayBlockingQueue<Event>> jobEventMap = new ConcurrentHashMap<>();
     /**
      * key: job id; <br>
      * value: job master;

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/disruptor/JobEventDisruptor.java

@@ -84,18 +84,13 @@ public void close() throws IOException {
         disruptor.shutdown();
     }
 
-    /** */
     public ArrayBlockingQueue<Event> storeJobHistory() {
         ArrayBlockingQueue<Event> events = new ArrayBlockingQueue<>(eventQueueSize);
         long nextSequence = 0;
         while (ringBuffer.getCursor() >= nextSequence) {
-            try {
-                JobEvent event = ringBuffer.get(nextSequence);
-                events.add(event.getEvent());
-                nextSequence++;
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
+            JobEvent event = ringBuffer.get(nextSequence);
+            events.add(event.getEvent());
+            nextSequence++;
         }
         try {
             this.close();

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/event/JobEventReportOperation.java

@@ -52,10 +52,7 @@ public void run() throws Exception {
         for (Event event : events) {
             processor.process(event);
             JobMaster jobMaster = coordinatorService.getJobMaster(Long.parseLong(event.getJobId()));
-            //            jobMaster.getEvents().add(event);
-            //            jobMaster.getHistoryEvents().add(event);
             jobMaster.getJobEventDisruptor().publish(event);
-            System.out.println("put 成功," + event);
         }
     }
 

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/operation/GetEventOperation.java

@@ -112,13 +112,9 @@ private void collectEvents(
             RingBuffer<JobEvent> ringBuffer, AtomicInteger sequence, List<Event> events) {
 
         while (ringBuffer.getCursor() >= sequence.get()) {
-            try {
-                JobEvent jobEvent = ringBuffer.get(sequence.get());
-                events.add(jobEvent.getEvent());
-                sequence.addAndGet(1);
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
+            JobEvent jobEvent = ringBuffer.get(sequence.get());
+            events.add(jobEvent.getEvent());
+            sequence.addAndGet(1);
         }
     }
 

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/rest/RestHttpGetCommandProcessor.java

@@ -23,7 +23,6 @@
 import org.apache.seatunnel.engine.server.NodeExtension;
 import org.apache.seatunnel.engine.server.log.FormatType;
 import org.apache.seatunnel.engine.server.log.Log4j2HttpGetCommandProcessor;
-import org.apache.seatunnel.engine.server.rest.service.EventService;
 import org.apache.seatunnel.engine.server.rest.service.JobInfoService;
 import org.apache.seatunnel.engine.server.rest.service.LogService;
 import org.apache.seatunnel.engine.server.rest.service.OverviewService;
@@ -55,7 +54,6 @@
 import static com.hazelcast.internal.ascii.rest.HttpStatusCode.SC_500;
 import static org.apache.seatunnel.engine.server.rest.RestConstant.CONTEXT_PATH;
 import static org.apache.seatunnel.engine.server.rest.RestConstant.INSTANCE_CONTEXT_PATH;
-import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_EVENT;
 import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_FINISHED_JOBS;
 import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_GET_ALL_LOG_NAME;
 import static org.apache.seatunnel.engine.server.rest.RestConstant.REST_URL_JOB_INFO;
@@ -81,7 +79,6 @@ public class RestHttpGetCommandProcessor extends HttpCommandProcessor<HttpGetCom
     private ThreadDumpService threadDumpService;
     private RunningThreadService runningThreadService;
     private LogService logService;
-    private EventService eventService;
 
     public RestHttpGetCommandProcessor(TextCommandService textCommandService) {
 
@@ -109,7 +106,6 @@ public RestHttpGetCommandProcessor(
         this.threadDumpService = new ThreadDumpService(nodeEngine);
         this.runningThreadService = new RunningThreadService(nodeEngine);
         this.logService = new LogService(nodeEngine);
-        this.eventService = new EventService(nodeEngine);
     }
 
     @Override
@@ -142,8 +138,6 @@ public void handle(HttpGetCommand httpGetCommand) {
                 getAllNodeLog(httpGetCommand, uri);
             } else if (uri.startsWith(CONTEXT_PATH + REST_URL_LOG)) {
                 getCurrentNodeLog(httpGetCommand, uri);
-            } else if (uri.startsWith(CONTEXT_PATH + REST_URL_EVENT)) {
-                getEventInfo(httpGetCommand);
             } else {
                 original.handle(httpGetCommand);
             }
@@ -164,33 +158,6 @@ public void handleRejection(HttpGetCommand httpGetCommand) {
         handle(httpGetCommand);
     }
 
-    public void getEventInfo(HttpGetCommand command) {
-        String uri = StringUtil.stripTrailingSlash(command.getURI());
-
-        int lastSlashIndex = uri.lastIndexOf('/');
-        if (lastSlashIndex == -1 || lastSlashIndex == uri.length() - 1) {
-            prepareResponse(SC_400, command, "Invalid event URI format");
-            return;
-        }
-
-        String jobIdStr = uri.substring(lastSlashIndex + 1);
-        if (StringUtils.isBlank(jobIdStr)) {
-            prepareResponse(SC_400, command, "Missing job ID in URI");
-            return;
-        }
-
-        try {
-            long jobId = Long.parseLong(jobIdStr);
-            this.prepareResponse(command, eventService.getEventInfoJson(jobId));
-        } catch (NumberFormatException e) {
-            logger.warning("Invalid job ID format: " + jobIdStr);
-            prepareResponse(SC_400, command, "Job ID must be a numeric value");
-        } catch (Exception e) {
-            logger.severe("Failed to get event info for job " + jobIdStr + ": " + e.getMessage());
-            prepareResponse(SC_500, command, "Internal server error");
-        }
-    }
-
     public void overView(HttpGetCommand command, String uri) {
         uri = StringUtil.stripTrailingSlash(uri);
         String tagStr;