backup

Author: zhangshenghang

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

@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.seatunnel.engine.server.disruptor;
 
 import org.apache.seatunnel.api.event.Event;

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

@@ -1,3 +1,20 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package org.apache.seatunnel.engine.server.disruptor;
 
 import org.apache.seatunnel.api.event.Event;

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

@@ -52,10 +52,10 @@ 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.getEvents().add(event);
+            //            jobMaster.getHistoryEvents().add(event);
             jobMaster.getJobEventDisruptor().publish(event);
-            System.out.println("put 成功,"+event);
+            System.out.println("put 成功," + event);
         }
     }
 

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobMaster.java

@@ -143,8 +143,6 @@ public class JobMaster {
 
     private SeaTunnelServer seaTunnelServer;
 
-    //    @Getter private ArrayBlockingQueue<Event> events;
-    //    @Getter private ArrayBlockingQueue<Event> historyEvents;
     @Getter private JobEventDisruptor jobEventDisruptor;
     /**
      * we need store slot used by task in Hazelcast IMap and release or reuse it when a new master
@@ -218,8 +216,6 @@ public JobMaster(
         this.metricsImap = metricsImap;
         this.seaTunnelServer = seaTunnelServer;
         this.releasedSlotWhenTaskGroupFinished = new ConcurrentHashMap<>();
-        //        this.events = new ArrayBlockingQueue<>(engineConfig.getEventQueueSize());
-        //        this.historyEvents = new ArrayBlockingQueue<>(engineConfig.getEventQueueSize());
         this.jobEventDisruptor = new JobEventDisruptor(engineConfig.getEventQueueSize());
     }
 

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

@@ -39,6 +39,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 @Slf4j
 public class GetEventOperation extends Operation implements IdentifiedDataSerializable {
@@ -47,13 +48,11 @@ public class GetEventOperation extends Operation implements IdentifiedDataSerial
 
     private Boolean isAll;
 
-    private List<Event> events = new ArrayList<>();
-
     public GetEventOperation() {}
 
     private Data response;
 
-    private long nextSequence;
+    private AtomicInteger nextSequence;
 
     /**
      * @param jobId job id
@@ -62,7 +61,7 @@ public GetEventOperation() {}
     public GetEventOperation(Long jobId, boolean isAll) {
         this.jobId = jobId;
         this.isAll = isAll;
-        this.nextSequence = 0;
+        this.nextSequence = new AtomicInteger(0);
     }
 
     @Override
@@ -93,35 +92,9 @@ private Data retrieveEvents(CoordinatorService coordinatorService) {
 
         if (jobMaster != null) {
             log.debug("Retrieving events for active job {}, isAll: {}", jobId, isAll);
-            if (isAll) {
-                RingBuffer<JobEvent> ringBuffer = jobMaster.getJobEventDisruptor().getRingBuffer();
-                long nextSequence = 0;
-                while (ringBuffer.getCursor() >= nextSequence) {
-                    try {
-                        JobEvent jobEvent = ringBuffer.get(nextSequence);
-                        events.add(jobEvent.getEvent());
-                        nextSequence++;
-                    } catch (Exception e) {
-                        e.printStackTrace();
-                    }
-                }
-                //
-                // Optional.ofNullable(jobMaster.getHistoryEvents()).ifPresent(events::addAll);
-            } else {
-                RingBuffer<JobEvent> ringBuffer = jobMaster.getJobEventDisruptor().getRingBuffer();
-                while (ringBuffer.getCursor() >= nextSequence) {
-                    try {
-                        JobEvent jobEvent = ringBuffer.get(nextSequence);
-                        System.out.println("add Event:" + jobEvent.getEvent());
-                        events.add(jobEvent.getEvent());
-                        nextSequence++;
-                    } catch (Exception e) {
-                        e.printStackTrace();
-                    }
-                }
-                //                Optional.ofNullable(jobMaster.getEvents()).ifPresent(q ->
-                // q.drainTo(events));
-            }
+            RingBuffer<JobEvent> ringBuffer = jobMaster.getJobEventDisruptor().getRingBuffer();
+            AtomicInteger sequenceToUse = isAll ? new AtomicInteger(0) : nextSequence;
+            collectEvents(ringBuffer, sequenceToUse, events);
         } else {
             log.debug("Job {} not active, retrieving from history", jobId);
             Optional.ofNullable(
@@ -135,6 +108,20 @@ private Data retrieveEvents(CoordinatorService coordinatorService) {
         return this.getNodeEngine().toData(events);
     }
 
+    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();
+            }
+        }
+    }
+
     @Override
     public Object getResponse() {
         return response;