GeneratorFrontType#nodePartition

Author: Nikita Sokolov

examples/src/main/java/com/spbsu/flamestream/example/nexmark/GeneratorFrontType.java

@@ -4,7 +4,6 @@
 import com.github.nexmark.flink.generator.GeneratorConfig;
 import com.github.nexmark.flink.generator.NexmarkGenerator;
 import com.spbsu.flamestream.core.data.PayloadDataItem;
-import com.spbsu.flamestream.core.data.meta.EdgeId;
 import com.spbsu.flamestream.core.data.meta.GlobalTime;
 import com.spbsu.flamestream.core.data.meta.Meta;
 import com.spbsu.flamestream.runtime.FlameRuntime;
@@ -20,18 +19,18 @@
 
 public class GeneratorFrontType implements FlameRuntime.FrontType<GeneratorFrontType.Front, Void> {
   private final NexmarkConfiguration nexmarkConfiguration;
-  private final Map<EdgeId, Integer> edgePartition;
+  private final Map<String, Integer> nodePartition;
   private final long baseTime;
   private final long maxEvents;
 
   public GeneratorFrontType(
           NexmarkConfiguration nexmarkConfiguration,
-          Map<EdgeId, Integer> edgePartition,
+          Map<String, Integer> nodePartition,
           long baseTime,
           long maxEvents
   ) {
     this.nexmarkConfiguration = nexmarkConfiguration;
-    this.edgePartition = edgePartition;
+    this.nodePartition = nodePartition;
     this.baseTime = baseTime;
     this.maxEvents = maxEvents;
   }
@@ -57,7 +56,7 @@ public static class Front implements com.spbsu.flamestream.runtime.edge.Front {
     public Front(EdgeContext edgeContext, GeneratorFrontType type) {
       this.edgeContext = edgeContext;
       this.type = type;
-      partition = type.edgePartition.get(edgeContext.edgeId());
+      partition = type.nodePartition.get(edgeContext.edgeId().nodeId());
     }
 
     @Override
@@ -73,7 +72,7 @@ public void onStart(Consumer<Object> consumer, GlobalTime from) {
               1,
               type.maxEvents,
               1
-      ).split(type.edgePartition.size()).get(partition);
+      ).split(type.nodePartition.size()).get(partition);
       final var executor =
               Executors.newSingleThreadExecutor(runnable -> new Thread(runnable, edgeContext.edgeId().toString()));
       executor.submit(() -> {

examples/src/test/java/com/spbsu/flamestream/example/nexmark/Query8Test.java

@@ -39,10 +39,10 @@ public void test() throws Exception {
                 flame.attachRear("rear", new TimingsRearType<>(new SimpleRearType(), 10)).collect(Collectors.toList());
 
         flame.attachFront("front", new GeneratorFrontType(nexmarkConfiguration, Map.ofEntries(
-                Map.entry(new EdgeId("front", "node-0"), 0),
-                Map.entry(new EdgeId("front", "node-1"), 1),
-                Map.entry(new EdgeId("front", "node-2"), 2),
-                Map.entry(new EdgeId("front", "node-3"), 3)
+                Map.entry("node-0", 0),
+                Map.entry("node-1", 1),
+                Map.entry("node-2", 2),
+                Map.entry("node-3", 3)
         ), currentTimeMillis, 10000));
         for (final var rear : rears) {
           rear.await();