You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by rp...@apache.org on 2021/10/03 09:36:03 UTC
[incubator-wayang] branch WAYANG-28 updated (df77465 -> 4c18b6d)
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a change to branch WAYANG-28
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git.
from df77465 [WAYANG-28] Shipper and Receiver with Rabbit and Kafka
new 8ea19f6 [WAYANG-28] General fixes
new 4c18b6d [WAYANG-28] Test
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
lolo | 121 ----
.../plugin/hackit/api/DataQuantaHackit.scala | 787 +++++++++++----------
.../apache/wayang/plugin/hackit/api/Hackit.scala | 18 +-
.../plugin/hackit/api/ApiExtensionTest.scala | 2 +-
.../wrapper/template/FlatMapTemplateSystem.java | 26 +-
.../wrapper/template/FunctionTemplateSystem.java | 29 +-
.../wayang/plugin/hackit/core/tags/PauseTag.java | 2 +-
.../wayang/plugin/hackit/core/tags/SkipTag.java | 2 +-
.../hackit/core/tuple/header/HeaderBuilder.java | 1 +
.../hackit/core/sniffer/HackitSnifferTest.java | 295 ++++++++
.../tagger/wrapper/PredicateWrapperHackitTest.java | 28 +
.../plugin/hackit/core/tuple/HackitTupleTest.java | 85 +++
.../core/tuple/header/HeaderBuilderTest.java | 33 +
.../hackit/core/tuple/header/HeaderLongTest.java | 24 +
.../hackit-shipper-rabbitmq/pom.xml | 18 +
.../rabbitmq/HachitShipperDirectRabbitMQ.java | 17 +
.../shipper/rabbitmq/HachitShipperRabbitMQ.java | 17 +
.../receiver/ReceiverMultiChannelRabbitMQ.java | 17 +
.../sender/SenderMultiChannelRabbitMQ.java | 17 +
.../src/main/resources/rabbitmq-config.properties | 16 +
.../hackit/shipper/kafka/sender/KafkaHackit.java | 17 +
.../src/main/resources/kafka-config.properties | 16 +
.../org/apache/wayang/hackit/sidecar/test.java | 17 +
.../hackit/sidecar/webservice/ExecutorManager.java | 17 +
.../hackit/sidecar/webservice/WebService.java | 17 +
.../src/main/resources/application.properties | 16 +
.../src/main/resources/claim.yaml | 16 +
.../src/main/resources/jobmanager-service.yaml | 16 +
.../src/main/resources/jobmanager.yaml | 16 +
.../src/main/resources/jobmanager_new.yaml | 16 +
.../src/main/resources/loadbalancer.yaml | 16 +
.../src/main/resources/taskmanager.yaml | 16 +
.../src/main/resources/taskmanager_new.yaml | 16 +
.../src/main/resources/volume.yaml | 16 +
34 files changed, 1261 insertions(+), 522 deletions(-)
delete mode 100644 lolo
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSnifferTest.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/PredicateWrapperHackitTest.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/HackitTupleTest.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilderTest.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderLongTest.java
[incubator-wayang] 02/02: [WAYANG-28] Test
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch WAYANG-28
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 4c18b6d423b42ae9c51d5c67de985d49fd73a27f
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Sun Oct 3 11:35:41 2021 +0200
[WAYANG-28] Test
---
.../wayang/plugin/hackit/core/tags/PauseTag.java | 2 +-
.../wayang/plugin/hackit/core/tags/SkipTag.java | 2 +-
.../hackit/core/tuple/header/HeaderBuilder.java | 1 +
.../hackit/core/sniffer/HackitSnifferTest.java | 295 +++++++++++++++++++++
.../tagger/wrapper/PredicateWrapperHackitTest.java | 28 ++
.../plugin/hackit/core/tuple/HackitTupleTest.java | 85 ++++++
.../core/tuple/header/HeaderBuilderTest.java | 33 +++
.../hackit/core/tuple/header/HeaderLongTest.java | 24 ++
8 files changed, 468 insertions(+), 2 deletions(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/PauseTag.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/PauseTag.java
index c347f21..4a7ef64 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/PauseTag.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/PauseTag.java
@@ -31,7 +31,7 @@ public class PauseTag extends HackitTag {
/**
* Default constructor
*/
- private PauseTag(){
+ public PauseTag(){
super();
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/SkipTag.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/SkipTag.java
index 9a1dbef..1366318 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/SkipTag.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tags/SkipTag.java
@@ -32,7 +32,7 @@ public class SkipTag extends HackitTag {
/**
* Default constructor
*/
- private SkipTag(){
+ public SkipTag(){
super();
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilder.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilder.java
index 9a94bf2..377e851 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilder.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilder.java
@@ -33,6 +33,7 @@ public class HeaderBuilder {
*/
public HeaderBuilder(){
configuration = new HashMap<>();
+ configuration.put("correlative.base", "1");
//TODO: take from the configuration
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSnifferTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSnifferTest.java
new file mode 100644
index 0000000..30791cc
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSnifferTest.java
@@ -0,0 +1,295 @@
+package org.apache.wayang.plugin.hackit.core.sniffer;
+
+import org.apache.wayang.plugin.hackit.core.sniffer.actor.Actor;
+import org.apache.wayang.plugin.hackit.core.sniffer.clone.BasicCloner;
+import org.apache.wayang.plugin.hackit.core.sniffer.inject.EmptyInjector;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.Shipper;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver.EmptyReceiver;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver.Receiver;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender.EmptySender;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender.Sender;
+import org.apache.wayang.plugin.hackit.core.sniffer.sniff.CollectionTagsToSniff;
+import org.apache.wayang.plugin.hackit.core.sniffer.sniff.SingleTagToSniff;
+import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple;
+import org.junit.Before;
+import org.junit.jupiter.api.Test;
+
+import java.util.Iterator;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class HackitSnifferTest {
+
+ @Test
+ void apply() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.apply(ht);
+ }
+
+ @Test
+ void setHackItInjector() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.setHackItInjector(new EmptyInjector<>());
+ }
+
+ @Test
+ void setActorFunction() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.setActorFunction(new Actor<HackitTuple<String, String>>() {
+ @Override
+ public boolean is_sendout(HackitTuple<String, String> value) {
+ return false;
+ }
+ });
+ }
+
+ @Test
+ void setShipper() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.setShipper(new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ });
+ }
+
+ @Test
+ void setHackItSniff() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.setHackItSniff(new CollectionTagsToSniff());
+ }
+
+ @Test
+ void setHackItCloner() {
+ HackitTuple<String, String> ht = new HackitTuple<>("test");
+ HackitSniffer hs = new HackitSniffer<String, String, byte[],
+ Sender<byte[]>, Receiver<HackitTuple<String, String>>>(
+ new EmptyInjector<>(),
+ null,
+ new Shipper() {
+ @Override
+ protected Sender createSenderInstance() {
+ return new MockUpSender();
+ }
+
+ @Override
+ protected Receiver createReceiverInstance() {
+ return new MockUpReceiver();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return false;
+ }
+
+ @Override
+ public Object next() {
+ return null;
+ }
+
+ },
+ new SingleTagToSniff(),
+ new BasicCloner<>()
+ );
+ hs.setHackItCloner(new BasicCloner<>());
+ }
+
+ class MockUpSender<T> implements Sender<T>{
+
+ @Override
+ public void init() {
+
+ }
+
+ @Override
+ public void send(T value) {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+ }
+
+ class MockUpReceiver<T> extends Receiver<T> {
+
+ @Override
+ public void init() {
+
+ }
+
+ @Override
+ public Iterator<T> getElements() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/PredicateWrapperHackitTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/PredicateWrapperHackitTest.java
new file mode 100644
index 0000000..0991622
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/PredicateWrapperHackitTest.java
@@ -0,0 +1,28 @@
+package org.apache.wayang.plugin.hackit.core.tagger.wrapper;
+
+import org.apache.wayang.plugin.hackit.core.tagger.wrapper.template.FunctionTemplateSystem;
+import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple;
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+class PredicateWrapperHackitTest {
+
+ @Test
+ public void testPredicateWrapper(){
+
+ FunctionTemplateSystem<String, Boolean> func = new FunctionTemplateSystem<String, Boolean>() {
+ @Override
+ public Boolean execute(String input) {
+ return input.length() > 6;
+ }
+ };
+
+ PredicateWrapperHackit<String, String> wrap = new PredicateWrapperHackit<String, String>(func);
+
+ String value = "test";
+ HackitTuple<String, String> tuple = new HackitTuple(value);
+ Boolean answer = wrap.execute(tuple);
+ Assert.assertTrue(answer instanceof Boolean);
+ }
+
+}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/HackitTupleTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/HackitTupleTest.java
new file mode 100644
index 0000000..effdbf7
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/HackitTupleTest.java
@@ -0,0 +1,85 @@
+package org.apache.wayang.plugin.hackit.core.tuple;
+
+import org.apache.wayang.plugin.hackit.core.tags.HackitTag;
+import org.apache.wayang.plugin.hackit.core.tags.LogTag;
+import org.apache.wayang.plugin.hackit.core.tags.PauseTag;
+import org.apache.wayang.plugin.hackit.core.tags.SkipTag;
+import org.apache.wayang.plugin.hackit.core.tuple.header.Header;
+import org.apache.wayang.plugin.hackit.core.tuple.header.HeaderLong;
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class HackitTupleTest {
+
+ @Test
+ void getHeader() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ Header<Long> header = ht.getHeader();
+ Assert.assertTrue(header instanceof Header);
+ Assert.assertTrue(header.getId() instanceof Long);
+ }
+
+ @Test
+ void getKey() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ Long k = ht.getKey();
+ Assert.assertTrue(k instanceof Long);
+ }
+
+ @Test
+ void getValue() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ String valueReturned = ht.getValue();
+ Assert.assertEquals(value, valueReturned);
+ }
+
+ @Test
+ void addTag() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ HackitTag logTag = new LogTag();
+ ht.addTag(logTag);
+ for (Iterator<HackitTag> it = ht.getTags(); it.hasNext(); ) {
+ HackitTag t = it.next();
+ Assert.assertTrue(logTag.equals(t));
+ }
+
+ }
+
+ @Test
+ void testAddTag() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ HackitTag tag1 = new LogTag();
+ HackitTag tag2 = new PauseTag();
+ HackitTag tag3 = new SkipTag();
+ Set<HackitTag> tags = new HashSet<>();
+ tags.add(tag1); tags.add(tag3); tags.add(tag2);
+ ht.addTag(tags);
+ for (Iterator<HackitTag> it = ht.getTags(); it.hasNext(); ) {
+ HackitTag t = it.next();
+ Assert.assertTrue(tags.contains(t));
+ }
+ }
+
+ @Test
+ void getTags() {
+ String value = "test";
+ HackitTuple<Long, String> ht = new HackitTuple<Long, String>(value);
+ HackitTag logTag = new LogTag();
+ ht.addTag(logTag);
+ for (Iterator<HackitTag> it = ht.getTags(); it.hasNext(); ) {
+ HackitTag t = it.next();
+ Assert.assertTrue(logTag.equals(t));
+ }
+ }
+}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilderTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilderTest.java
new file mode 100644
index 0000000..80e70e5
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderBuilderTest.java
@@ -0,0 +1,33 @@
+package org.apache.wayang.plugin.hackit.core.tuple.header;
+
+import org.assertj.core.api.MapAssert;
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class HeaderBuilderTest {
+
+ @Test
+ void generateHeader() {
+ HeaderBuilder h = new HeaderBuilder();
+ Header<Object> hh = h.generateHeader();
+ }
+
+ @Test
+ void getConfiguration() {
+ HeaderBuilder h = new HeaderBuilder();
+ for (Map.Entry<String, String> l : h.getConfiguration().entrySet()) {
+ Assert.assertTrue(l instanceof Map.Entry);
+ }
+ }
+
+ @Test
+ void setConfiguration() {
+ HeaderBuilder h = new HeaderBuilder();
+ h.setConfiguration("test", "test");
+ }
+
+}
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderLongTest.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderLongTest.java
new file mode 100644
index 0000000..62f0546
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/test/java/org/apache/wayang/plugin/hackit/core/tuple/header/HeaderLongTest.java
@@ -0,0 +1,24 @@
+package org.apache.wayang.plugin.hackit.core.tuple.header;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class HeaderLongTest {
+
+ @Test
+ void createChild() {
+ HeaderLong l = new HeaderLong(1l);
+ HeaderLong ll = l.createChild();
+ Assert.assertTrue(ll.getId() instanceof Long);
+ }
+
+ @Test
+ void generateID() {
+ HeaderLong l = new HeaderLong(1l);
+ Long ll = l.generateID();
+ Assert.assertTrue(ll instanceof Long);
+
+ }
+}
\ No newline at end of file
[incubator-wayang] 01/02: [WAYANG-28] General fixes
Posted by rp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch WAYANG-28
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 8ea19f6d62567d01ce06e961cbec1208427c29eb
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Fri Oct 1 10:59:04 2021 +0200
[WAYANG-28] General fixes
---
lolo | 121 ----
.../plugin/hackit/api/DataQuantaHackit.scala | 787 +++++++++++----------
.../apache/wayang/plugin/hackit/api/Hackit.scala | 18 +-
.../plugin/hackit/api/ApiExtensionTest.scala | 2 +-
.../wrapper/template/FlatMapTemplateSystem.java | 26 +-
.../wrapper/template/FunctionTemplateSystem.java | 29 +-
.../hackit-shipper-rabbitmq/pom.xml | 18 +
.../rabbitmq/HachitShipperDirectRabbitMQ.java | 17 +
.../shipper/rabbitmq/HachitShipperRabbitMQ.java | 17 +
.../receiver/ReceiverMultiChannelRabbitMQ.java | 17 +
.../sender/SenderMultiChannelRabbitMQ.java | 17 +
.../src/main/resources/rabbitmq-config.properties | 16 +
.../hackit/shipper/kafka/sender/KafkaHackit.java | 17 +
.../src/main/resources/kafka-config.properties | 16 +
.../org/apache/wayang/hackit/sidecar/test.java | 17 +
.../hackit/sidecar/webservice/ExecutorManager.java | 17 +
.../hackit/sidecar/webservice/WebService.java | 17 +
.../src/main/resources/application.properties | 16 +
.../src/main/resources/claim.yaml | 16 +
.../src/main/resources/jobmanager-service.yaml | 16 +
.../src/main/resources/jobmanager.yaml | 16 +
.../src/main/resources/jobmanager_new.yaml | 16 +
.../src/main/resources/loadbalancer.yaml | 16 +
.../src/main/resources/taskmanager.yaml | 16 +
.../src/main/resources/taskmanager_new.yaml | 16 +
.../src/main/resources/volume.yaml | 16 +
26 files changed, 793 insertions(+), 520 deletions(-)
diff --git a/lolo b/lolo
deleted file mode 100644
index 11a7c99..0000000
--- a/lolo
+++ /dev/null
@@ -1,121 +0,0 @@
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/bin/java -ea -Didea.test.cyclic.buffer.size=1048576 -javaagent
-/Users/bertty/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/211.7142.45/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=61718
-/Users/bertty/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/211.7142.45/IntelliJ IDEA.app/Contents/bin -Dfile.encoding=UTF-8 -classpath /Users/bertty/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/211.7142.45/IntelliJ IDEA.app/Contents/lib/idea_rt.jar
-/Users/bertty/.m2/repository/org/junit/platform/junit-platform-launcher/1.6.1/junit-platform-launcher-1.6.1.jar
-/Users/bertty/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/211.7142.45/IntelliJ IDEA.app/Contents/plugins/junit/lib/junit5-rt.jar
-/Users/bertty/Library/Application Support/JetBrains/Toolbox/apps/IDEA-U/ch-0/211.7142.45/IntelliJ IDEA.app/Contents/plugins/junit/lib/junit-rt.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/charsets.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/cldrdata.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/dnsns.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/jaccess.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/jfxrt.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/localedata.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/nashorn.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/sunec.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/sunjce_provider.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/sunpkcs11.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/ext/zipfs.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/jce.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/jfxswt.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/jsse.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/management-agent.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/resources.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/jre/lib/rt.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/ant-javafx.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/dt.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/javafx-mx.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/jconsole.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/packager.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/sa-jdi.jar
-/Users/bertty/Library/Java/JavaVirtualMachines/corretto-1.8.0_252/Contents/Home/lib/tools.jar
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-plugins/wayang-hackit/wayang-hackit-api/target/test-classes
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-plugins/wayang-hackit/wayang-hackit-api/target/classes
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-api/wayang-api-scala-java/target/classes
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-commons/wayang-core/target/classes
-/Users/bertty/.m2/repository/org/json/json/20160212/json-20160212.jar
-/Users/bertty/.m2/repository/org/yaml/snakeyaml/1.17/snakeyaml-1.17.jar
-/Users/bertty/.m2/repository/de/odysseus/juel/juel-api/2.2.7/juel-api-2.2.7.jar
-/Users/bertty/.m2/repository/de/odysseus/juel/juel-impl/2.2.7/juel-impl-2.2.7.jar
-/Users/bertty/.m2/repository/de/hpi/isg/profiledb-instrumentation/0.1.1/profiledb-instrumentation-0.1.1.jar
-/Users/bertty/.m2/repository/org/apache/commons/commons-lang3/3.3.2/commons-lang3-3.3.2.jar
-/Users/bertty/.m2/repository/org/apache/logging/log4j/log4j-api/2.14.0/log4j-api-2.14.0.jar
-/Users/bertty/.m2/repository/org/apache/logging/log4j/log4j-core/2.14.0/log4j-core-2.14.0.jar
-/Users/bertty/.m2/repository/net/sf/trove4j/trove4j/3.0.3/trove4j-3.0.3.jar
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-commons/wayang-basic/target/classes
-/Users/bertty/.m2/repository/de/hpi/isg/profiledb-store/0.1.1/profiledb-store-0.1.1.jar
-/Users/bertty/IdeaProjects/incubator-wayang/wayang-platforms/wayang-java/target/classes
-/Users/bertty/.m2/repository/org/scala-lang/scala-library/2.11.12/scala-library-2.11.12.jar
-/Users/bertty/.m2/repository/org/apache/hadoop/hadoop-common/2.7.7/hadoop-common-2.7.7.jar
-/Users/bertty/.m2/repository/org/apache/hadoop/hadoop-annotations/2.7.7/hadoop-annotations-2.7.7.jar
-/Users/bertty/.m2/repository/com/google/guava/guava/11.0.2/guava-11.0.2.jar
-/Users/bertty/.m2/repository/commons-cli/commons-cli/1.2/commons-cli-1.2.jar
-/Users/bertty/.m2/repository/org/apache/commons/commons-math3/3.1.1/commons-math3-3.1.1.jar
-/Users/bertty/.m2/repository/xmlenc/xmlenc/0.52/xmlenc-0.52.jar
-/Users/bertty/.m2/repository/commons-httpclient/commons-httpclient/3.1/commons-httpclient-3.1.jar
-/Users/bertty/.m2/repository/commons-codec/commons-codec/1.4/commons-codec-1.4.jar
-/Users/bertty/.m2/repository/commons-io/commons-io/2.4/commons-io-2.4.jar
-/Users/bertty/.m2/repository/commons-net/commons-net/3.1/commons-net-3.1.jar
-/Users/bertty/.m2/repository/commons-collections/commons-collections/3.2.2/commons-collections-3.2.2.jar
-/Users/bertty/.m2/repository/org/mortbay/jetty/jetty-sslengine/6.1.26/jetty-sslengine-6.1.26.jar
-/Users/bertty/.m2/repository/commons-logging/commons-logging/1.1.3/commons-logging-1.1.3.jar
-/Users/bertty/.m2/repository/log4j/log4j/1.2.17/log4j-1.2.17.jar
-/Users/bertty/.m2/repository/commons-lang/commons-lang/2.6/commons-lang-2.6.jar
-/Users/bertty/.m2/repository/commons-configuration/commons-configuration/1.6/commons-configuration-1.6.jar
-/Users/bertty/.m2/repository/commons-digester/commons-digester/1.8/commons-digester-1.8.jar
-/Users/bertty/.m2/repository/commons-beanutils/commons-beanutils/1.7.0/commons-beanutils-1.7.0.jar
-/Users/bertty/.m2/repository/commons-beanutils/commons-beanutils-core/1.8.0/commons-beanutils-core-1.8.0.jar
-/Users/bertty/.m2/repository/org/slf4j/slf4j-api/1.7.10/slf4j-api-1.7.10.jar
-/Users/bertty/.m2/repository/org/slf4j/slf4j-log4j12/1.7.10/slf4j-log4j12-1.7.10.jar
-/Users/bertty/.m2/repository/org/codehaus/jackson/jackson-core-asl/1.9.13/jackson-core-asl-1.9.13.jar
-/Users/bertty/.m2/repository/org/codehaus/jackson/jackson-mapper-asl/1.9.13/jackson-mapper-asl-1.9.13.jar
-/Users/bertty/.m2/repository/org/apache/avro/avro/1.7.4/avro-1.7.4.jar
-/Users/bertty/.m2/repository/com/thoughtworks/paranamer/paranamer/2.3/paranamer-2.3.jar
-/Users/bertty/.m2/repository/org/xerial/snappy/snappy-java/1.0.4.1/snappy-java-1.0.4.1.jar
-/Users/bertty/.m2/repository/com/google/protobuf/protobuf-java/2.5.0/protobuf-java-2.5.0.jar
-/Users/bertty/.m2/repository/com/google/code/gson/gson/2.2.4/gson-2.2.4.jar
-/Users/bertty/.m2/repository/org/apache/hadoop/hadoop-auth/2.7.7/hadoop-auth-2.7.7.jar
-/Users/bertty/.m2/repository/org/apache/httpcomponents/httpclient/4.2.5/httpclient-4.2.5.jar
-/Users/bertty/.m2/repository/org/apache/httpcomponents/httpcore/4.2.4/httpcore-4.2.4.jar
-/Users/bertty/.m2/repository/org/apache/directory/server/apacheds-kerberos-codec/2.0.0-M15/apacheds-kerberos-codec-2.0.0-M15.jar
-/Users/bertty/.m2/repository/org/apache/directory/server/apacheds-i18n/2.0.0-M15/apacheds-i18n-2.0.0-M15.jar
-/Users/bertty/.m2/repository/org/apache/directory/api/api-asn1-api/1.0.0-M20/api-asn1-api-1.0.0-M20.jar
-/Users/bertty/.m2/repository/org/apache/directory/api/api-util/1.0.0-M20/api-util-1.0.0-M20.jar
-/Users/bertty/.m2/repository/org/apache/curator/curator-framework/2.7.1/curator-framework-2.7.1.jar
-/Users/bertty/.m2/repository/org/apache/curator/curator-client/2.7.1/curator-client-2.7.1.jar
-/Users/bertty/.m2/repository/org/apache/curator/curator-recipes/2.7.1/curator-recipes-2.7.1.jar
-/Users/bertty/.m2/repository/com/google/code/findbugs/jsr305/3.0.0/jsr305-3.0.0.jar
-/Users/bertty/.m2/repository/org/apache/htrace/htrace-core/3.1.0-incubating/htrace-core-3.1.0-incubating.jar
-/Users/bertty/.m2/repository/org/apache/zookeeper/zookeeper/3.4.6/zookeeper-3.4.6.jar
-/Users/bertty/.m2/repository/org/apache/commons/commons-compress/1.4.1/commons-compress-1.4.1.jar
-/Users/bertty/.m2/repository/org/tukaani/xz/1.0/xz-1.0.jar
-/Users/bertty/.m2/repository/org/apache/hadoop/hadoop-hdfs/2.7.7/hadoop-hdfs-2.7.7.jar
-/Users/bertty/.m2/repository/org/mortbay/jetty/jetty-util/6.1.26/jetty-util-6.1.26.jar
-/Users/bertty/.m2/repository/io/netty/netty/3.6.2.Final/netty-3.6.2.Final.jar
-/Users/bertty/.m2/repository/io/netty/netty-all/4.0.23.Final/netty-all-4.0.23.Final.jar
-/Users/bertty/.m2/repository/xerces/xercesImpl/2.9.1/xercesImpl-2.9.1.jar
-/Users/bertty/.m2/repository/xml-apis/xml-apis/1.3.04/xml-apis-1.3.04.jar
-/Users/bertty/.m2/repository/org/fusesource/leveldbjni/leveldbjni-all/1.8/leveldbjni-all-1.8.jar
-/Users/bertty/.m2/repository/org/junit/jupiter/junit-jupiter/5.6.1/junit-jupiter-5.6.1.jar
-/Users/bertty/.m2/repository/org/junit/jupiter/junit-jupiter-api/5.6.1/junit-jupiter-api-5.6.1.jar
-/Users/bertty/.m2/repository/org/opentest4j/opentest4j/1.2.0/opentest4j-1.2.0.jar
-/Users/bertty/.m2/repository/org/junit/platform/junit-platform-commons/1.6.1/junit-platform-commons-1.6.1.jar
-/Users/bertty/.m2/repository/org/junit/jupiter/junit-jupiter-params/5.6.1/junit-jupiter-params-5.6.1.jar
-/Users/bertty/.m2/repository/org/junit/jupiter/junit-jupiter-engine/5.6.1/junit-jupiter-engine-5.6.1.jar
-/Users/bertty/.m2/repository/org/junit/vintage/junit-vintage-engine/5.6.1/junit-vintage-engine-5.6.1.jar
-/Users/bertty/.m2/repository/org/apiguardian/apiguardian-api/1.1.0/apiguardian-api-1.1.0.jar
-/Users/bertty/.m2/repository/org/junit/platform/junit-platform-engine/1.6.1/junit-platform-engine-1.6.1.jar
-/Users/bertty/.m2/repository/junit/junit/4.12/junit-4.12.jar
-/Users/bertty/.m2/repository/org/hamcrest/hamcrest-core/1.3/hamcrest-core-1.3.jar
-/Users/bertty/.m2/repository/org/mockito/mockito-core/3.5.10/mockito-core-3.5.10.jar
-/Users/bertty/.m2/repository/net/bytebuddy/byte-buddy/1.10.13/byte-buddy-1.10.13.jar
-/Users/bertty/.m2/repository/net/bytebuddy/byte-buddy-agent/1.10.13/byte-buddy-agent-1.10.13.jar
-/Users/bertty/.m2/repository/org/objenesis/objenesis/3.1/objenesis-3.1.jar
-/Users/bertty/.m2/repository/org/mockito/mockito-junit-jupiter/3.5.10/mockito-junit-jupiter-3.5.10.jar
-/Users/bertty/.m2/repository/org/assertj/assertj-core/3.17.2/assertj-core-3.17.2.jar
-/Users/bertty/.m2/repository/org/antlr/antlr4/4.9.1/antlr4-4.9.1.jar
-/Users/bertty/.m2/repository/org/antlr/antlr4-runtime/4.9.1/antlr4-runtime-4.9.1.jar
-/Users/bertty/.m2/repository/org/antlr/antlr-runtime/3.5.2/antlr-runtime-3.5.2.jar
-/Users/bertty/.m2/repository/org/antlr/ST4/4.3/ST4-4.3.jar
-/Users/bertty/.m2/repository/org/abego/treelayout/org.abego.treelayout.core/1.0.3/org.abego.treelayout.core-1.0.3.jar
-/Users/bertty/.m2/repository/org/glassfish/javax.json/1.0.4/javax.json-1.0.4.jar
-/Users/bertty/.m2/repository/com/ibm/icu/icu4j/61.1/icu4j-61.1.jar com.intellij.rt.junit.JUnitStarter -ideVersion5 -junit5 org.apache.wayang.api.ApiTest,testReadMapCollect
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/DataQuantaHackit.scala b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/DataQuantaHackit.scala
index f1505e0..4d7a49f 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/DataQuantaHackit.scala
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/DataQuantaHackit.scala
@@ -49,401 +49,402 @@ import scala.reflect.ClassTag
* @param ev$1 the data type of the elements in this instance
* @param planBuilder keeps track of the [[WayangPlan]] being build
*/
-class DataQuantaHackit[Out : ClassTag]
- (override val operator: ElementaryOperator, outputIndex: Int = 0)
- (implicit override val planBuilder: PlanBuilder)
- extends DataQuanta[Out](operator, outputIndex)
-{
-
- var tagger : HackitTagger = null;
- //var sniffer : HackitSniffer = null;
-
- /**
- * add a [[HackitTag]] on the [[HackitTagger]] to enable the process
- *
- * @param tag [[HackitTag]] to be added
- * @return the self instance
- */
- //TODO add the version of collection
- def addTag(tag: HackitTag): DataQuantaHackit[Out] = {
-// tagger.addPreTag(tag)
- println("here")
- this
- }
-
- def underHackit[Key: ClassTag](): DataQuantaHackit[HackitTuple[Key, Out]] = {
-// HackItRDD<K, T> ktHackItRDD = new HackItRDD<K, T>(
-// rdd.map((Function<T, HackItTuple<K, T>>) HackItTuple::new)
-// );
-
-null
- }
-
- /**
- * Feed this instance into a [[MapOperator]].
- *
- * @param udf a Java 8 lambda expression as UDF for the [[MapOperator]]
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[MapOperator]]'s output
- */
- override def mapJava[NewOut: ClassTag](udf: SerializableFunction[Out, NewOut],
- udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
-
- val wrapper_internal = new FunctionWrapperHackit[Object, Out, NewOut](udf)
- this.tagger = wrapper_internal
-
- val wrapper = new SerializableFunction[HackitTuple[Object, Out], HackitTuple[Object, NewOut]](){
- override def apply(t: HackitTuple[Object, Out]): HackitTuple[Object, NewOut] = {
- println(t)
- wrapper_internal.apply(t)
- }
- }
-
- val mapOperator = new MapOperator(new TransformationDescriptor(
- wrapper, basicDataUnitType[HackitTuple[Object, Out]], basicDataUnitType[HackitTuple[Object, NewOut]], udfLoad
- ))
- this.connectTo(mapOperator, 0)
- DataQuantaHackit.wrap[NewOut](mapOperator)
- }
-
- /**
- * Feed this instance into a [[MapPartitionsOperator]].
- *
- * @param udf a Java 8 lambda expression as UDF for the [[MapPartitionsOperator]]
- * @param selectivity selectivity of the UDF
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[MapOperator]]'s output
- */
- override def mapPartitionsJava[NewOut: ClassTag](udf: SerializableFunction[JavaIterable[Out], JavaIterable[NewOut]],
- selectivity: ProbabilisticDoubleInterval = null,
- udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
- val mapOperator = new MapPartitionsOperator(
- new MapPartitionsDescriptor(udf, basicDataUnitType[Out], basicDataUnitType[NewOut], selectivity, udfLoad)
- )
- this.connectTo(mapOperator, 0)
- DataQuantaHackit.wrap[NewOut](mapOperator)
- }
-
- /**
- * Feed this instance into a [[MapOperator]] with a [[ProjectionDescriptor]].
- *
- * @param fieldNames names of the fields to be projected
- * @return a new instance representing the [[MapOperator]]'s output
- */
- override def project[NewOut: ClassTag](fieldNames: Seq[String]): DataQuantaHackit[NewOut] = {
- val projectionOperator = new MapOperator(
- new ProjectionDescriptor(basicDataUnitType[Out], basicDataUnitType[NewOut], fieldNames: _*)
- )
- this.connectTo(projectionOperator, 0)
- DataQuantaHackit.wrap[NewOut](projectionOperator)
- }
-
- /**
- * Feed this instance into a [[FilterOperator]].
- *
- * @param udf UDF for the [[FilterOperator]]
- * @param sqlUdf UDF as SQL `WHERE` clause
- * @param selectivity selectivity of the UDF
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[FilterOperator]]'s output
- */
- override def filterJava(udf: SerializablePredicate[Out],
- sqlUdf: String = null,
- selectivity: ProbabilisticDoubleInterval = null,
- udfLoad: LoadProfileEstimator = null): DataQuantaHackit[Out] = {
-// val filterOperator = new FilterOperator(new PredicateDescriptor(
-// udf, this.output.getType.getDataUnitType.toBasicDataUnitType, selectivity, udfLoad
-// ).withSqlImplementation(sqlUdf))
-// this.connectTo(filterOperator, 0)
-// DataQuantaHackit.wrap[Out](filterOperator)
- null
- }
-
- /**
- * Feed this instance into a [[FlatMapOperator]].
- *
- * @param udf a Java 8 lambda expression as UDF for the [[FlatMapOperator]]
- * @param selectivity selectivity of the UDF
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[FlatMapOperator]]'s output
- */
- override def flatMapJava[NewOut: ClassTag](udf: SerializableFunction[Out, JavaIterable[NewOut]],
- selectivity: ProbabilisticDoubleInterval = null,
- udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
-// val flatMapOperator = new FlatMapOperator(new FlatMapDescriptor(
-// udf, basicDataUnitType[Out], basicDataUnitType[NewOut], selectivity, udfLoad
+//TODO: validate this class its possible that need to be removed
+class DataQuantaHackit {//[Out : ClassTag]
+// (override val operator: ElementaryOperator, outputIndex: Int = 0)
+// (implicit override val planBuilder: PlanBuilder)
+// extends DataQuanta[Out](operator, outputIndex)
+//{
+
+// var tagger : HackitTagger = null;
+// //var sniffer : HackitSniffer = null;
+//
+// /**
+// * add a [[HackitTag]] on the [[HackitTagger]] to enable the process
+// *
+// * @param tag [[HackitTag]] to be added
+// * @return the self instance
+// */
+// //TODO add the version of collection
+// def addTag(tag: HackitTag): DataQuantaHackit[Out] = {
+//// tagger.addPreTag(tag)
+// println("here")
+// this
+// }
+//
+// def underHackit[Key: ClassTag](): DataQuantaHackit[HackitTuple[Key, Out]] = {
+//// HackItRDD<K, T> ktHackItRDD = new HackItRDD<K, T>(
+//// rdd.map((Function<T, HackItTuple<K, T>>) HackItTuple::new)
+//// );
+//
+//null
+// }
+//
+// /**
+// * Feed this instance into a [[MapOperator]].
+// *
+// * @param udf a Java 8 lambda expression as UDF for the [[MapOperator]]
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[MapOperator]]'s output
+// */
+// override def mapJava[NewOut: ClassTag](udf: SerializableFunction[Out, NewOut],
+// udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
+//
+// val wrapper_internal = new FunctionWrapperHackit[Object, Out, NewOut](udf)
+// this.tagger = wrapper_internal
+//
+// val wrapper = new SerializableFunction[HackitTuple[Object, Out], HackitTuple[Object, NewOut]](){
+// override def apply(t: HackitTuple[Object, Out]): HackitTuple[Object, NewOut] = {
+// println(t)
+// wrapper_internal.apply(t)
+// }
+// }
+//
+// val mapOperator = new MapOperator(new TransformationDescriptor(
+// wrapper, basicDataUnitType[HackitTuple[Object, Out]], basicDataUnitType[HackitTuple[Object, NewOut]], udfLoad
// ))
-// this.connectTo(flatMapOperator, 0)
-// DataQuantaHackit.wrap[NewOut](flatMapOperator)
- null
- }
-
-
- /**
- * Feed this instance into a [[SampleOperator]].
- *
- * @param sampleSizeFunction absolute size of the sample as a function of the current iteration number
- * @param datasetSize optional size of the dataset to be sampled
- * @param sampleMethod the [[SampleOperator.Methods]] to use for sampling
- * @return a new instance representing the [[FlatMapOperator]]'s output
- */
- override def sampleDynamicJava(sampleSizeFunction: IntUnaryOperator,
- datasetSize: Long = SampleOperator.UNKNOWN_DATASET_SIZE,
- seed: Option[Long] = None,
- sampleMethod: SampleOperator.Methods = SampleOperator.Methods.ANY): DataQuantaHackit[Out] = {
- if (seed.isEmpty) {
- val sampleOperator = new SampleOperator(
- sampleSizeFunction,
- dataSetType[Out],
- sampleMethod
- )
- sampleOperator.setDatasetSize(datasetSize)
- this.connectTo(sampleOperator, 0)
- DataQuantaHackit.wrap[Out](sampleOperator)
- }
- else {
- val sampleOperator = new SampleOperator(
- sampleSizeFunction,
- dataSetType[Out],
- sampleMethod,
- seed.get
- )
- sampleOperator.setDatasetSize(datasetSize)
- this.connectTo(sampleOperator, 0)
- DataQuantaHackit.wrap[Out](sampleOperator)
- }
- }
-
- /**
- * Assigns this instance a key extractor, which enables some key-based operations.
- *
- * @see KeyedDataQuanta
- * @param keyExtractor extracts the key from the [[DataQuantaDefault]]
- * @return the [[KeyedDataQuanta]]
- */
- //TODO validate this implementation
- override def keyByJava[Key: ClassTag](keyExtractor: SerializableFunction[Out, Key]) : KeyedDataQuanta[Out, Key] = {
- new KeyedDataQuanta[Out, Key](this, keyExtractor)
- }
-
- /**
- * Feed this instance into a [[ReduceByOperator]].
- *
- * @param keyUdf UDF to extract the grouping key from the data quanta
- * @param udf aggregation UDF for the [[ReduceByOperator]]
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[ReduceByOperator]]'s output
- */
- override def reduceByKeyJava[Key: ClassTag](keyUdf: SerializableFunction[Out, Key],
- udf: SerializableBinaryOperator[Out],
- udfLoad: LoadProfileEstimator = null)
- : DataQuantaHackit[Out] = {
-// val reduceByOperator = new ReduceByOperator(
-// new TransformationDescriptor(keyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
-// new ReduceDescriptor(udf, groupedDataUnitType[Out], basicDataUnitType[Out], udfLoad)
-// )
-// this.connectTo(reduceByOperator, 0)
-// DataQuantaHackit.wrap[Out](reduceByOperator)
- null
- }
-
- /**
- * Feed this instance into a [[MaterializedGroupByOperator]].
- *
- * @param keyUdf UDF to extract the grouping key from the data quanta
- * @param keyUdfLoad optional [[LoadProfileEstimator]] for the `keyUdf`
- * @return a new instance representing the [[MaterializedGroupByOperator]]'s output
- */
- override def groupByKeyJava[Key: ClassTag](keyUdf: SerializableFunction[Out, Key],
- keyUdfLoad: LoadProfileEstimator = null): DataQuantaHackit[java.lang.Iterable[Out]] = {
-// val groupByOperator = new MaterializedGroupByOperator(
-// new TransformationDescriptor(keyUdf, basicDataUnitType[Out], basicDataUnitType[Key], keyUdfLoad),
-// dataSetType[Out],
-// groupedDataSetType[Out]
-// )
-// this.connectTo(groupByOperator, 0)
-// DataQuantaHackit.wrap[java.lang.Iterable[Out]](groupByOperator)
- null
- }
-
- /**
- * Feed this instance into a [[GlobalReduceOperator]].
- *
- * @param udf aggregation UDF for the [[GlobalReduceOperator]]
- * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
- * @return a new instance representing the [[GlobalReduceOperator]]'s output
- */
- override def reduceJava(udf: SerializableBinaryOperator[Out],
- udfLoad: LoadProfileEstimator = null): DataQuantaHackit[Out] = {
-// val globalReduceOperator = new GlobalReduceOperator(
-// new ReduceDescriptor(udf, groupedDataUnitType[Out], basicDataUnitType[Out], udfLoad)
+// this.connectTo(mapOperator, 0)
+// DataQuantaHackit.wrap[NewOut](mapOperator)
+// }
+//
+// /**
+// * Feed this instance into a [[MapPartitionsOperator]].
+// *
+// * @param udf a Java 8 lambda expression as UDF for the [[MapPartitionsOperator]]
+// * @param selectivity selectivity of the UDF
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[MapOperator]]'s output
+// */
+// override def mapPartitionsJava[NewOut: ClassTag](udf: SerializableFunction[JavaIterable[Out], JavaIterable[NewOut]],
+// selectivity: ProbabilisticDoubleInterval = null,
+// udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
+// val mapOperator = new MapPartitionsOperator(
+// new MapPartitionsDescriptor(udf, basicDataUnitType[Out], basicDataUnitType[NewOut], selectivity, udfLoad)
// )
-// this.connectTo(globalReduceOperator, 0)
-// DataQuantaHackit.wrap[Out](globalReduceOperator)
- null
- }
-
- /**
- * Feeds this and a further instance into a [[JoinOperator]].
- *
- * @param thisKeyUdf UDF to extract keys from data quanta in this instance
- * @param that the other instance
- * @param thatKeyUdf UDF to extract keys from data quanta from `that` instance
- * @return a new instance representing the [[JoinOperator]]'s output
- */
- override def joinJava[ThatOut: ClassTag, Key: ClassTag](thisKeyUdf: SerializableFunction[Out, Key], that: DataQuanta[ThatOut], thatKeyUdf: SerializableFunction[ThatOut, Key]): DataQuantaHackit[WayangTuple2[Out, ThatOut]] = {
-// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
-// val joinOperator = new JoinOperator(
-// new TransformationDescriptor(thisKeyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
-// new TransformationDescriptor(thatKeyUdf, basicDataUnitType[ThatOut], basicDataUnitType[Key])
+// this.connectTo(mapOperator, 0)
+// DataQuantaHackit.wrap[NewOut](mapOperator)
+// }
+//
+// /**
+// * Feed this instance into a [[MapOperator]] with a [[ProjectionDescriptor]].
+// *
+// * @param fieldNames names of the fields to be projected
+// * @return a new instance representing the [[MapOperator]]'s output
+// */
+// override def project[NewOut: ClassTag](fieldNames: Seq[String]): DataQuantaHackit[NewOut] = {
+// val projectionOperator = new MapOperator(
+// new ProjectionDescriptor(basicDataUnitType[Out], basicDataUnitType[NewOut], fieldNames: _*)
// )
-// this.connectTo(joinOperator, 0)
-// that.connectTo(joinOperator, 1)
-// DataQuantaHackit.wrap[WayangTuple2[Out, ThatOut]](joinOperator)
- null
- }
-
- /**
- * Feeds this and a further instance into a [[CoGroupOperator]].
- *
- * @param thisKeyUdf UDF to extract keys from data quanta in this instance
- * @param that the other instance
- * @param thatKeyUdf UDF to extract keys from data quanta from `that` instance
- * @return a new instance representing the [[CoGroupOperator]]'s output
- */
- override def coGroupJava[ThatOut: ClassTag, Key: ClassTag](thisKeyUdf: SerializableFunction[Out, Key], that: DataQuanta[ThatOut], thatKeyUdf: SerializableFunction[ThatOut, Key]): DataQuantaHackit[WayangTuple2[JavaIterable[Out], JavaIterable[ThatOut]]] = {
-// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
-// val coGroupOperator = new CoGroupOperator(
-// new TransformationDescriptor(thisKeyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
-// new TransformationDescriptor(thatKeyUdf, basicDataUnitType[ThatOut], basicDataUnitType[Key])
-// )
-// this.connectTo(coGroupOperator, 0)
-// that.connectTo(coGroupOperator, 1)
-// DataQuantaHackit.wrap[WayangTuple2[java.lang.Iterable[Out], java.lang.Iterable[ThatOut]]](coGroupOperator)
- null
- }
-
-
- /**
- * Feeds this and a further instance into a [[SortOperator]].
- *
- * @param keyUdf UDF to extract key from data quanta in this instance
- * @return a new instance representing the [[SortOperator]]'s output
- */
- override def sortJava[Key: ClassTag] (keyUdf: SerializableFunction[Out, Key]) : DataQuantaHackit[Out] = {
-// val sortOperator = new SortOperator(new TransformationDescriptor(
-// keyUdf, basicDataUnitType[Out], basicDataUnitType[Key]))
-// this.connectTo(sortOperator, 0)
-// DataQuantaHackit.wrap[Out](sortOperator)
- null
- }
-
-
- /**
- * Feed this instance into a [[GlobalMaterializedGroupOperator]].
- *
- * @return a new instance representing the [[GlobalMaterializedGroupOperator]]'s output
- */
- override def group(): DataQuantaHackit[JavaIterable[Out]] = {
-// val groupOperator = new GlobalMaterializedGroupOperator(dataSetType[Out], groupedDataSetType[Out])
-// this.connectTo(groupOperator, 0)
-// DataQuantaHackit.wrap[JavaIterable[Out]](groupOperator)
- null
- }
-
- /**
- * Feed this instance and a further instance into a [[UnionAllOperator]].
- *
- * @param that the other instance to union with
- * @return a new instance representing the [[UnionAllOperator]]'s output
- */
- override def union(that: DataQuanta[Out]): DataQuantaHackit[Out] = {
-// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
-// val unionAllOperator = new UnionAllOperator(dataSetType[Out])
-// this.connectTo(unionAllOperator, 0)
-// that.connectTo(unionAllOperator, 1)
-// DataQuantaHackit.wrap[Out](unionAllOperator)
- null
- }
-
- /**
- * Feed this instance and a further instance into a [[IntersectOperator]].
- *
- * @param that the other instance to intersect with
- * @return a new instance representing the [[IntersectOperator]]'s output
- */
- override def intersect(that: DataQuanta[Out]): DataQuantaHackit[Out] = {
-// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
-// val intersectOperator = new IntersectOperator(dataSetType[Out])
-// this.connectTo(intersectOperator, 0)
-// that.connectTo(intersectOperator, 1)
-// DataQuantaHackit.wrap[Out](intersectOperator)
- null
- }
-
- /**
- * Feeds this and a further instance into a [[CartesianOperator]].
- *
- * @param that the other instance
- * @return a new instance representing the [[CartesianOperator]]'s output
- */
- override def cartesian[ThatOut: ClassTag](that: DataQuanta[ThatOut]): DataQuantaHackit[WayangTuple2[Out, ThatOut]] = {
-// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
-// val cartesianOperator = new CartesianOperator(dataSetType[Out], dataSetType[ThatOut])
-// this.connectTo(cartesianOperator, 0)
-// that.connectTo(cartesianOperator, 1)
-// DataQuantaHackit.wrap[WayangTuple2[Out, ThatOut]](cartesianOperator)
- null
- }
-
- /**
- * Feeds this instance into a [[ZipWithIdOperator]].
- *
- * @return a new instance representing the [[ZipWithIdOperator]]'s output
- */
- override def zipWithId: DataQuantaHackit[WayangTuple2[lang.Long, Out]] = {
-// val zipWithIdOperator = new ZipWithIdOperator(dataSetType[Out])
-// this.connectTo(zipWithIdOperator, 0)
-// DataQuantaHackit.wrap[WayangTuple2[lang.Long, Out]](zipWithIdOperator)
- null
- }
-
- /**
- * Feeds this instance into a [[DistinctOperator]].
- *
- * @return a new instance representing the [[DistinctOperator]]'s output
- */
- override def distinct: DataQuantaHackit[Out] = {
-// val distinctOperator = new DistinctOperator(dataSetType[Out])
-// this.connectTo(distinctOperator, 0)
-// DataQuantaHackit.wrap[Out](distinctOperator)
- null
- }
-
- /**
- * Feeds this instance into a [[CountOperator]].
- *
- * @return a new instance representing the [[CountOperator]]'s output
- */
- override def count: DataQuantaHackit[lang.Long] = {
-// val countOperator = new CountOperator(dataSetType[Out])
-// this.connectTo(countOperator, 0)
-// DataQuantaHackit.wrap[lang.Long](countOperator)
- null
- }
+// this.connectTo(projectionOperator, 0)
+// DataQuantaHackit.wrap[NewOut](projectionOperator)
+// }
+//
+// /**
+// * Feed this instance into a [[FilterOperator]].
+// *
+// * @param udf UDF for the [[FilterOperator]]
+// * @param sqlUdf UDF as SQL `WHERE` clause
+// * @param selectivity selectivity of the UDF
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[FilterOperator]]'s output
+// */
+// override def filterJava(udf: SerializablePredicate[Out],
+// sqlUdf: String = null,
+// selectivity: ProbabilisticDoubleInterval = null,
+// udfLoad: LoadProfileEstimator = null): DataQuantaHackit[Out] = {
+//// val filterOperator = new FilterOperator(new PredicateDescriptor(
+//// udf, this.output.getType.getDataUnitType.toBasicDataUnitType, selectivity, udfLoad
+//// ).withSqlImplementation(sqlUdf))
+//// this.connectTo(filterOperator, 0)
+//// DataQuantaHackit.wrap[Out](filterOperator)
+// null
+// }
+//
+// /**
+// * Feed this instance into a [[FlatMapOperator]].
+// *
+// * @param udf a Java 8 lambda expression as UDF for the [[FlatMapOperator]]
+// * @param selectivity selectivity of the UDF
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[FlatMapOperator]]'s output
+// */
+// override def flatMapJava[NewOut: ClassTag](udf: SerializableFunction[Out, JavaIterable[NewOut]],
+// selectivity: ProbabilisticDoubleInterval = null,
+// udfLoad: LoadProfileEstimator = null): DataQuantaHackit[NewOut] = {
+//// val flatMapOperator = new FlatMapOperator(new FlatMapDescriptor(
+//// udf, basicDataUnitType[Out], basicDataUnitType[NewOut], selectivity, udfLoad
+//// ))
+//// this.connectTo(flatMapOperator, 0)
+//// DataQuantaHackit.wrap[NewOut](flatMapOperator)
+// null
+// }
+//
+//
+// /**
+// * Feed this instance into a [[SampleOperator]].
+// *
+// * @param sampleSizeFunction absolute size of the sample as a function of the current iteration number
+// * @param datasetSize optional size of the dataset to be sampled
+// * @param sampleMethod the [[SampleOperator.Methods]] to use for sampling
+// * @return a new instance representing the [[FlatMapOperator]]'s output
+// */
+// override def sampleDynamicJava(sampleSizeFunction: IntUnaryOperator,
+// datasetSize: Long = SampleOperator.UNKNOWN_DATASET_SIZE,
+// seed: Option[Long] = None,
+// sampleMethod: SampleOperator.Methods = SampleOperator.Methods.ANY): DataQuantaHackit[Out] = {
+// if (seed.isEmpty) {
+// val sampleOperator = new SampleOperator(
+// sampleSizeFunction,
+// dataSetType[Out],
+// sampleMethod
+// )
+// sampleOperator.setDatasetSize(datasetSize)
+// this.connectTo(sampleOperator, 0)
+// DataQuantaHackit.wrap[Out](sampleOperator)
+// }
+// else {
+// val sampleOperator = new SampleOperator(
+// sampleSizeFunction,
+// dataSetType[Out],
+// sampleMethod,
+// seed.get
+// )
+// sampleOperator.setDatasetSize(datasetSize)
+// this.connectTo(sampleOperator, 0)
+// DataQuantaHackit.wrap[Out](sampleOperator)
+// }
+// }
+//
+// /**
+// * Assigns this instance a key extractor, which enables some key-based operations.
+// *
+// * @see KeyedDataQuanta
+// * @param keyExtractor extracts the key from the [[DataQuantaDefault]]
+// * @return the [[KeyedDataQuanta]]
+// */
+// //TODO validate this implementation
+// override def keyByJava[Key: ClassTag](keyExtractor: SerializableFunction[Out, Key]) : KeyedDataQuanta[Out, Key] = {
+// new KeyedDataQuanta[Out, Key](this, keyExtractor)
+// }
+//
+// /**
+// * Feed this instance into a [[ReduceByOperator]].
+// *
+// * @param keyUdf UDF to extract the grouping key from the data quanta
+// * @param udf aggregation UDF for the [[ReduceByOperator]]
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[ReduceByOperator]]'s output
+// */
+// override def reduceByKeyJava[Key: ClassTag](keyUdf: SerializableFunction[Out, Key],
+// udf: SerializableBinaryOperator[Out],
+// udfLoad: LoadProfileEstimator = null)
+// : DataQuantaHackit[Out] = {
+//// val reduceByOperator = new ReduceByOperator(
+//// new TransformationDescriptor(keyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
+//// new ReduceDescriptor(udf, groupedDataUnitType[Out], basicDataUnitType[Out], udfLoad)
+//// )
+//// this.connectTo(reduceByOperator, 0)
+//// DataQuantaHackit.wrap[Out](reduceByOperator)
+// null
+// }
+//
+// /**
+// * Feed this instance into a [[MaterializedGroupByOperator]].
+// *
+// * @param keyUdf UDF to extract the grouping key from the data quanta
+// * @param keyUdfLoad optional [[LoadProfileEstimator]] for the `keyUdf`
+// * @return a new instance representing the [[MaterializedGroupByOperator]]'s output
+// */
+// override def groupByKeyJava[Key: ClassTag](keyUdf: SerializableFunction[Out, Key],
+// keyUdfLoad: LoadProfileEstimator = null): DataQuantaHackit[java.lang.Iterable[Out]] = {
+//// val groupByOperator = new MaterializedGroupByOperator(
+//// new TransformationDescriptor(keyUdf, basicDataUnitType[Out], basicDataUnitType[Key], keyUdfLoad),
+//// dataSetType[Out],
+//// groupedDataSetType[Out]
+//// )
+//// this.connectTo(groupByOperator, 0)
+//// DataQuantaHackit.wrap[java.lang.Iterable[Out]](groupByOperator)
+// null
+// }
+//
+// /**
+// * Feed this instance into a [[GlobalReduceOperator]].
+// *
+// * @param udf aggregation UDF for the [[GlobalReduceOperator]]
+// * @param udfLoad optional [[LoadProfileEstimator]] for the `udf`
+// * @return a new instance representing the [[GlobalReduceOperator]]'s output
+// */
+// override def reduceJava(udf: SerializableBinaryOperator[Out],
+// udfLoad: LoadProfileEstimator = null): DataQuantaHackit[Out] = {
+//// val globalReduceOperator = new GlobalReduceOperator(
+//// new ReduceDescriptor(udf, groupedDataUnitType[Out], basicDataUnitType[Out], udfLoad)
+//// )
+//// this.connectTo(globalReduceOperator, 0)
+//// DataQuantaHackit.wrap[Out](globalReduceOperator)
+// null
+// }
+//
+// /**
+// * Feeds this and a further instance into a [[JoinOperator]].
+// *
+// * @param thisKeyUdf UDF to extract keys from data quanta in this instance
+// * @param that the other instance
+// * @param thatKeyUdf UDF to extract keys from data quanta from `that` instance
+// * @return a new instance representing the [[JoinOperator]]'s output
+// */
+// override def joinJava[ThatOut: ClassTag, Key: ClassTag](thisKeyUdf: SerializableFunction[Out, Key], that: DataQuanta[ThatOut], thatKeyUdf: SerializableFunction[ThatOut, Key]): DataQuantaHackit[WayangTuple2[Out, ThatOut]] = {
+//// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
+//// val joinOperator = new JoinOperator(
+//// new TransformationDescriptor(thisKeyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
+//// new TransformationDescriptor(thatKeyUdf, basicDataUnitType[ThatOut], basicDataUnitType[Key])
+//// )
+//// this.connectTo(joinOperator, 0)
+//// that.connectTo(joinOperator, 1)
+//// DataQuantaHackit.wrap[WayangTuple2[Out, ThatOut]](joinOperator)
+// null
+// }
+//
+// /**
+// * Feeds this and a further instance into a [[CoGroupOperator]].
+// *
+// * @param thisKeyUdf UDF to extract keys from data quanta in this instance
+// * @param that the other instance
+// * @param thatKeyUdf UDF to extract keys from data quanta from `that` instance
+// * @return a new instance representing the [[CoGroupOperator]]'s output
+// */
+// override def coGroupJava[ThatOut: ClassTag, Key: ClassTag](thisKeyUdf: SerializableFunction[Out, Key], that: DataQuanta[ThatOut], thatKeyUdf: SerializableFunction[ThatOut, Key]): DataQuantaHackit[WayangTuple2[JavaIterable[Out], JavaIterable[ThatOut]]] = {
+//// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
+//// val coGroupOperator = new CoGroupOperator(
+//// new TransformationDescriptor(thisKeyUdf, basicDataUnitType[Out], basicDataUnitType[Key]),
+//// new TransformationDescriptor(thatKeyUdf, basicDataUnitType[ThatOut], basicDataUnitType[Key])
+//// )
+//// this.connectTo(coGroupOperator, 0)
+//// that.connectTo(coGroupOperator, 1)
+//// DataQuantaHackit.wrap[WayangTuple2[java.lang.Iterable[Out], java.lang.Iterable[ThatOut]]](coGroupOperator)
+// null
+// }
+//
+//
+// /**
+// * Feeds this and a further instance into a [[SortOperator]].
+// *
+// * @param keyUdf UDF to extract key from data quanta in this instance
+// * @return a new instance representing the [[SortOperator]]'s output
+// */
+// override def sortJava[Key: ClassTag] (keyUdf: SerializableFunction[Out, Key]) : DataQuantaHackit[Out] = {
+//// val sortOperator = new SortOperator(new TransformationDescriptor(
+//// keyUdf, basicDataUnitType[Out], basicDataUnitType[Key]))
+//// this.connectTo(sortOperator, 0)
+//// DataQuantaHackit.wrap[Out](sortOperator)
+// null
+// }
+//
+//
+// /**
+// * Feed this instance into a [[GlobalMaterializedGroupOperator]].
+// *
+// * @return a new instance representing the [[GlobalMaterializedGroupOperator]]'s output
+// */
+// override def group(): DataQuantaHackit[JavaIterable[Out]] = {
+//// val groupOperator = new GlobalMaterializedGroupOperator(dataSetType[Out], groupedDataSetType[Out])
+//// this.connectTo(groupOperator, 0)
+//// DataQuantaHackit.wrap[JavaIterable[Out]](groupOperator)
+// null
+// }
+//
+// /**
+// * Feed this instance and a further instance into a [[UnionAllOperator]].
+// *
+// * @param that the other instance to union with
+// * @return a new instance representing the [[UnionAllOperator]]'s output
+// */
+// override def union(that: DataQuanta[Out]): DataQuantaHackit[Out] = {
+//// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
+//// val unionAllOperator = new UnionAllOperator(dataSetType[Out])
+//// this.connectTo(unionAllOperator, 0)
+//// that.connectTo(unionAllOperator, 1)
+//// DataQuantaHackit.wrap[Out](unionAllOperator)
+// null
+// }
+//
+// /**
+// * Feed this instance and a further instance into a [[IntersectOperator]].
+// *
+// * @param that the other instance to intersect with
+// * @return a new instance representing the [[IntersectOperator]]'s output
+// */
+// override def intersect(that: DataQuanta[Out]): DataQuantaHackit[Out] = {
+//// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
+//// val intersectOperator = new IntersectOperator(dataSetType[Out])
+//// this.connectTo(intersectOperator, 0)
+//// that.connectTo(intersectOperator, 1)
+//// DataQuantaHackit.wrap[Out](intersectOperator)
+// null
+// }
+//
+// /**
+// * Feeds this and a further instance into a [[CartesianOperator]].
+// *
+// * @param that the other instance
+// * @return a new instance representing the [[CartesianOperator]]'s output
+// */
+// override def cartesian[ThatOut: ClassTag](that: DataQuanta[ThatOut]): DataQuantaHackit[WayangTuple2[Out, ThatOut]] = {
+//// require(this.planBuilder eq that.planBuilder, s"$this and $that must use the same plan builders.")
+//// val cartesianOperator = new CartesianOperator(dataSetType[Out], dataSetType[ThatOut])
+//// this.connectTo(cartesianOperator, 0)
+//// that.connectTo(cartesianOperator, 1)
+//// DataQuantaHackit.wrap[WayangTuple2[Out, ThatOut]](cartesianOperator)
+// null
+// }
+//
+// /**
+// * Feeds this instance into a [[ZipWithIdOperator]].
+// *
+// * @return a new instance representing the [[ZipWithIdOperator]]'s output
+// */
+// override def zipWithId: DataQuantaHackit[WayangTuple2[lang.Long, Out]] = {
+//// val zipWithIdOperator = new ZipWithIdOperator(dataSetType[Out])
+//// this.connectTo(zipWithIdOperator, 0)
+//// DataQuantaHackit.wrap[WayangTuple2[lang.Long, Out]](zipWithIdOperator)
+// null
+// }
+//
+// /**
+// * Feeds this instance into a [[DistinctOperator]].
+// *
+// * @return a new instance representing the [[DistinctOperator]]'s output
+// */
+// override def distinct: DataQuantaHackit[Out] = {
+//// val distinctOperator = new DistinctOperator(dataSetType[Out])
+//// this.connectTo(distinctOperator, 0)
+//// DataQuantaHackit.wrap[Out](distinctOperator)
+// null
+// }
+//
+// /**
+// * Feeds this instance into a [[CountOperator]].
+// *
+// * @return a new instance representing the [[CountOperator]]'s output
+// */
+// override def count: DataQuantaHackit[lang.Long] = {
+//// val countOperator = new CountOperator(dataSetType[Out])
+//// this.connectTo(countOperator, 0)
+//// DataQuantaHackit.wrap[lang.Long](countOperator)
+// null
+// }
}
-object DataQuantaHackit extends DataQuantaCreator{
-
- def wrap[T:ClassTag](operator: ElementaryOperator, outputIndex: Int = 0)(implicit planBuilder: PlanBuilder): DataQuantaHackit[T] = {
- new DataQuantaHackit[T](operator, outputIndex)
- }
-
- def create[T](output: OutputSlot[T])(implicit planBuilder: PlanBuilder): DataQuantaHackit[_] =
- new DataQuantaHackit(output.getOwner.asInstanceOf[ElementaryOperator], output.getIndex)(ClassTag(output.getType.getDataUnitType.getTypeClass), planBuilder)
-
- implicit def wrapDataQuanta[T:ClassTag](dataQuanta: DataQuanta[T]): DataQuantaHackit[T] = {
- new DataQuantaHackit[T](dataQuanta.operator, dataQuanta.outputIndex)(ClassTag(dataQuanta.output.getType.getDataUnitType.getTypeClass), dataQuanta.planBuilder);
- }
-
-}
+//object DataQuantaHackit extends DataQuantaCreator{
+//
+// def wrap[T:ClassTag](operator: ElementaryOperator, outputIndex: Int = 0)(implicit planBuilder: PlanBuilder): DataQuantaHackit[T] = {
+// new DataQuantaHackit[T](operator, outputIndex)
+// }
+//
+// def create[T](output: OutputSlot[T])(implicit planBuilder: PlanBuilder): DataQuantaHackit[_] =
+// new DataQuantaHackit(output.getOwner.asInstanceOf[ElementaryOperator], output.getIndex)(ClassTag(output.getType.getDataUnitType.getTypeClass), planBuilder)
+//
+// implicit def wrapDataQuanta[T:ClassTag](dataQuanta: DataQuanta[T]): DataQuantaHackit[T] = {
+// new DataQuantaHackit[T](dataQuanta.operator, dataQuanta.outputIndex)(ClassTag(dataQuanta.output.getType.getDataUnitType.getTypeClass), dataQuanta.planBuilder);
+// }
+//
+//}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/Hackit.scala b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/Hackit.scala
index b9499fb..2b47a55 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/Hackit.scala
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/main/scala/org/apache/wayang/plugin/hackit/api/Hackit.scala
@@ -28,6 +28,7 @@ import org.apache.wayang.plugin.hackit.core.tagger.wrapper.FunctionWrapperHackit
import org.apache.wayang.plugin.hackit.core.tags.HackitTag
import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple
import org.apache.wayang.api.toSerializableFunction
+import org.apache.wayang.plugin.hackit.core.tagger.wrapper.template.FunctionTemplateSystem
import scala.language.implicitConversions
import scala.reflect.{ClassTag, classTag}
@@ -59,11 +60,22 @@ class Hackit[Key: ClassTag, Type:ClassTag](implicit var planBuilder: PlanBuilder
def map[TypeOut: ClassTag](udf: Type => TypeOut, udfLoad: LoadProfileEstimator = null): Hackit[Key, TypeOut] = {
val hackit = new Hackit[Key, TypeOut]()
- val wrapper = new FunctionWrapperHackit[Key, Type, TypeOut](toSerializableFunction(udf))
+ val wrapper = toWayangFunction(udf)
hackit.dataQuanta = dataQuanta.map[HackitTuple[Key, TypeOut]](element => wrapper.apply(element), udfLoad)
hackit
}
+ def toWayangFunction[In, Out](scalaFunc: In => Out):SerializableFunction[HackitTuple[Key, In], HackitTuple[Key, Out]] =
+ new SerializableFunction[HackitTuple[Key, In], HackitTuple[Key, Out]] {
+ val fun = new FunctionWrapperHackit[Key, In, Out](toHackitFunction(scalaFunc))
+ override def apply(t: HackitTuple[Key, In]): HackitTuple[Key, Out] = fun(t)
+ }
+
+
+ def toHackitFunction[In, Out](scalaFunc: In => Out): FunctionTemplateSystem[In, Out] =
+ new FunctionTemplateSystem[In, Out] {
+ override def execute(t: In) = scalaFunc(t)
+ }
}
@@ -71,8 +83,10 @@ object Hackit {
//TODO: replace the object with a parameter
implicit def underHackit[T: ClassTag](dataQuanta: DataQuanta[T]): Hackit[java.lang.Object, T] = {
- return new Hackit[java.lang.Object, T]()(classTag[java.lang.Object], ClassTag(dataQuanta.output.getType.getDataUnitType.getTypeClass), dataQuanta.planBuilder)
+ return new Hackit[java.lang.Object, T]()(classTag[java.lang.Object], ClassTag(dataQuanta.output.getType.getDataUnitType.getTypeClass) : ClassTag[T], dataQuanta.planBuilder)
.underHackit(dataQuanta)
}
+
+
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/test/scala/org/apache/wayang/plugin/hackit/api/ApiExtensionTest.scala b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/test/scala/org/apache/wayang/plugin/hackit/api/ApiExtensionTest.scala
index 42867f3..3d92309 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-api/src/test/scala/org/apache/wayang/plugin/hackit/api/ApiExtensionTest.scala
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-api/src/test/scala/org/apache/wayang/plugin/hackit/api/ApiExtensionTest.scala
@@ -33,7 +33,7 @@ class ApiExtensionTest {
@BeforeEach
def setUp() ={
- DataQuantaFactory.setTemplate(DataQuantaHackit);
+// DataQuantaFactory.setTemplate(DataQuantaHackit);
}
@Test
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FlatMapTemplateSystem.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FlatMapTemplateSystem.java
index 8f3b6fc..1a4974f 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FlatMapTemplateSystem.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FlatMapTemplateSystem.java
@@ -1,8 +1,32 @@
+/*
+ * 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.wayang.plugin.hackit.core.tagger.wrapper.template;
+import java.io.Serializable;
import java.util.Iterator;
+import java.util.function.Function;
-public interface FlatMapTemplateSystem<I, O> {
+public interface FlatMapTemplateSystem<I, O> extends Function<I, Iterator<O>>, Serializable {
public Iterator<O> execute(I input);
+
+ @Override
+ default Iterator<O> apply(I i) {
+ return this.execute(i);
+ }
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FunctionTemplateSystem.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FunctionTemplateSystem.java
index f19f8cc..2e62472 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FunctionTemplateSystem.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/tagger/wrapper/template/FunctionTemplateSystem.java
@@ -1,6 +1,31 @@
+/*
+ * 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.wayang.plugin.hackit.core.tagger.wrapper.template;
-public interface FunctionTemplateSystem<I, O> {
+import java.io.Serializable;
+import java.util.function.Function;
- public O execute(I input);
+public interface FunctionTemplateSystem<I, O> extends Function<I, O>, Serializable {
+
+ O execute(I input);
+
+ @Override
+ default O apply(I i) {
+ return execute(i);
+ }
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/pom.xml b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/pom.xml
index b3220bd..4f7d561 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/pom.xml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/pom.xml
@@ -1,4 +1,22 @@
<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+ -->
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperDirectRabbitMQ.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperDirectRabbitMQ.java
index 046fcd7..c8dac80 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperDirectRabbitMQ.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperDirectRabbitMQ.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.wayang.hackit.shipper.rabbitmq;
import com.rabbitmq.client.Channel;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperRabbitMQ.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperRabbitMQ.java
index 93ba2b1..5c814e7 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperRabbitMQ.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/HachitShipperRabbitMQ.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.wayang.hackit.shipper.rabbitmq;
public class HachitShipperRabbitMQ {
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/receiver/ReceiverMultiChannelRabbitMQ.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/receiver/ReceiverMultiChannelRabbitMQ.java
index bff40cb..58b333d 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/receiver/ReceiverMultiChannelRabbitMQ.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/receiver/ReceiverMultiChannelRabbitMQ.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.wayang.hackit.shipper.rabbitmq.receiver;
import com.rabbitmq.client.*;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/sender/SenderMultiChannelRabbitMQ.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/sender/SenderMultiChannelRabbitMQ.java
index 3a098dd..c1f34a6 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/sender/SenderMultiChannelRabbitMQ.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/java/org/apache/wayang/hackit/shipper/rabbitmq/sender/SenderMultiChannelRabbitMQ.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.wayang.hackit.shipper.rabbitmq.sender;
import com.rabbitmq.client.Channel;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/resources/rabbitmq-config.properties b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/resources/rabbitmq-config.properties
index c326cdc..27faa95 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/resources/rabbitmq-config.properties
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/hackit-shipper-rabbitmq/src/main/resources/rabbitmq-config.properties
@@ -1,3 +1,19 @@
+#
+# 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.
+#
username=admin
password=admin
virtualhost=/
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/KafkaHackit.java b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/KafkaHackit.java
index 6a3e1c9..2ef08a0 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/KafkaHackit.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/java/org/apache/wayang/hackit/shipper/kafka/sender/KafkaHackit.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.wayang.hackit.shipper.kafka.sender;
import org.apache.commons.lang3.SerializationUtils;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/resources/kafka-config.properties b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/resources/kafka-config.properties
index e69de29..85f7634 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/resources/kafka-config.properties
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-shipper/wayang-hackit-shipper-kafka/src/main/resources/kafka-config.properties
@@ -0,0 +1,16 @@
+#
+# 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.
+#
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/test.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/test.java
index 816384a..5a38413 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/test.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/test.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.wayang.hackit.sidecar;
import org.apache.flink.api.common.functions.FilterFunction;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java
index f3bd83a..2b2b07e 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/ExecutorManager.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.wayang.hackit.sidecar.webservice;
import java.io.BufferedReader;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
index ec605b4..04be8ad 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/java/org/apache/wayang/hackit/sidecar/webservice/WebService.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.wayang.hackit.sidecar.webservice;
import org.apache.flink.api.common.JobExecutionResult;
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/application.properties b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/application.properties
index 8d51d0c..4a088d9 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/application.properties
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/application.properties
@@ -1 +1,17 @@
+#
+# 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.
+#
server.port=8082
\ No newline at end of file
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/claim.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/claim.yaml
index f1de882..beffe30 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/claim.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/claim.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
index a3c235a..48d5ea3 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager-service.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: v1
kind: Service
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
index 6919861..49baee3 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: apps/v1
kind: Deployment
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager_new.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager_new.yaml
index 8ab2f48..1f0baeb 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager_new.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/jobmanager_new.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: apps/v1
kind: Deployment
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/loadbalancer.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/loadbalancer.yaml
index e0625f4..930bec5 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/loadbalancer.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/loadbalancer.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: v1
kind: Service
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
index 03e1ae8..ab565f3 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: apps/v1
kind: Deployment
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager_new.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager_new.yaml
index 6344a15..e5e1b10 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager_new.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/taskmanager_new.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: apps/v1
kind: Deployment
metadata:
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/volume.yaml b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/volume.yaml
index 9d328c1..be4c9ab 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/volume.yaml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-sidecar/src/main/resources/volume.yaml
@@ -1,3 +1,19 @@
+#
+# 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.
+#
apiVersion: v1
kind: PersistentVolume
metadata: