last

Author: Caideyipi

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java

@@ -393,7 +393,8 @@ public void testSourcePermission() {
         receiverEnv,
         "select count(*) from root.vehicle.**",
         "count(root.vehicle.car.temperature),",
-        Collections.singleton("1,"));
+        Collections.singleton("1,"),
+        20);
 
     // test showing pipe
     // Create another pipe, user is root

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/sync/IoTDBSchemaRegionSink.java

@@ -131,9 +131,7 @@ private void doTransfer(final PipeSchemaRegionWritePlanEvent pipeSchemaRegionWri
           true);
     }
 
-    if (LOGGER.isDebugEnabled()) {
-      LOGGER.debug("Successfully transferred schema event {}.", pipeSchemaRegionWritePlanEvent);
-    }
+    LOGGER.info("Successfully transferred schema event {}.", pipeSchemaRegionWritePlanEvent);
   }
 
   private void doTransferWrapper(final PipeSchemaRegionSnapshotEvent pipeSchemaRegionSnapshotEvent)

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/source/IoTDBNonDataRegionSource.java

@@ -68,6 +68,8 @@ public abstract class IoTDBNonDataRegionSource extends IoTDBSource {
   // the extractor is closed and then be reused by processor.
   protected final AtomicBoolean hasBeenClosed = new AtomicBoolean(false);
 
+  protected PipeWritePlanEvent lastEvent = null;
+
   protected abstract AbstractPipeListeningQueue getListeningQueue();
 
   @Override
@@ -163,7 +165,7 @@ public EnrichedEvent supply() throws Exception {
     }
 
     // Check whether snapshot being parsed exists
-    PipeWritePlanEvent realtimeEvent = null;
+    PipeWritePlanEvent realtimeEvent = lastEvent;
     if (hasNextEventInCurrentSnapshot()) {
       realtimeEvent = getNextEventInCurrentSnapshot();
     }
@@ -178,7 +180,7 @@ public EnrichedEvent supply() throws Exception {
                       pipeName,
                       creationTime,
                       pipeTaskMeta,
-                      (TreePattern) treePattern,
+                      treePattern,
                       tablePattern,
                       userId,
                       userName,
@@ -211,17 +213,17 @@ public EnrichedEvent supply() throws Exception {
 
     // Realtime
     if (Objects.isNull(realtimeEvent)) {
-      realtimeEvent = (PipeWritePlanEvent) iterator.peek(getMaxBlockingTimeMs());
+      realtimeEvent = (PipeWritePlanEvent) iterator.next(getMaxBlockingTimeMs());
     }
     if (Objects.isNull(realtimeEvent)) {
       return null;
     }
+    lastEvent = realtimeEvent;
 
     realtimeEvent =
         trimRealtimeEventByPipePattern(realtimeEvent)
             .flatMap(this::trimRealtimeEventByPrivilege)
             .orElse(null);
-    iterator.next(0);
 
     if (Objects.isNull(realtimeEvent)
         || !isTypeListened(realtimeEvent)
@@ -232,6 +234,7 @@ public EnrichedEvent supply() throws Exception {
         event.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
       }
       event.increaseReferenceCount(IoTDBNonDataRegionSource.class.getName());
+      lastEvent = null;
       return event;
     }
 
@@ -253,6 +256,7 @@ public EnrichedEvent supply() throws Exception {
       realtimeEvent.bindProgressIndex(new MetaProgressIndex(iterator.getNextIndex() - 1));
     }
     realtimeEvent.increaseReferenceCount(IoTDBNonDataRegionSource.class.getName());
+    lastEvent = null;
     return realtimeEvent;
   }