You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2016/02/20 09:56:47 UTC
[1/7] incubator-apex-malhar git commit: APEXMALHAR-1970 #resolve
#comment Fix the ArrayOutOfBoundaryException and add a bunch of tests for
both one_to_one and one_to_many partition
Repository: incubator-apex-malhar
Updated Branches:
refs/heads/release-3.3 6ad18e8ca -> 040f8f777
APEXMALHAR-1970 #resolve #comment Fix the ArrayOutOfBoundaryException and add a bunch of tests for both one_to_one and one_to_many partition
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/cd183db8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/cd183db8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/cd183db8
Branch: refs/heads/release-3.3
Commit: cd183db8f4b6427dedb1bdda869c2a6b7042f430
Parents: 6ad18e8
Author: Siyuan Hua <hs...@apache.org>
Authored: Mon Jan 11 22:09:25 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:13:02 2016 -0800
----------------------------------------------------------------------
.../malhar/kafka/AbstractKafkaPartitioner.java | 4 ++--
.../apex/malhar/kafka/OneToManyPartitioner.java | 4 ++--
.../malhar/kafka/KafkaInputOperatorTest.java | 23 ++++++++++++--------
3 files changed, 18 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cd183db8/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
index e01c38e..c708145 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
@@ -259,13 +259,13 @@ public abstract class AbstractKafkaPartitioner implements Partitioner<AbstractKa
}
PartitionMeta that = (PartitionMeta)o;
return Objects.equals(cluster, that.cluster) &&
- Objects.equals(topicPartition, that.topicPartition);
+ Objects.equals(getTopicPartition(), that.getTopicPartition());
}
@Override
public int hashCode()
{
- return Objects.hash(cluster, topicPartition);
+ return Objects.hash(cluster, getTopicPartition());
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cd183db8/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
index 16f4da4..bcd3073 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
@@ -54,8 +54,8 @@ public class OneToManyPartitioner extends AbstractKafkaPartitioner
for (Map.Entry<String, List<PartitionInfo>> topicPartition : clusterMap.getValue().entrySet()) {
for (PartitionInfo pif : topicPartition.getValue()) {
int index = i++ % partitionCount;
- if (eachPartitionAssignment.get(index) == null) {
- eachPartitionAssignment.add(index, new HashSet<PartitionMeta>());
+ if (index >= eachPartitionAssignment.size()) {
+ eachPartitionAssignment.add(new HashSet<PartitionMeta>());
}
eachPartitionAssignment.get(index).add(new PartitionMeta(clusterMap.getKey(), topicPartition.getKey(), pif.partition()));
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cd183db8/kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java
----------------------------------------------------------------------
diff --git a/kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java b/kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java
index 17bc465..d055555 100644
--- a/kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java
+++ b/kafka/src/test/java/org/apache/apex/malhar/kafka/KafkaInputOperatorTest.java
@@ -48,26 +48,30 @@ public class KafkaInputOperatorTest extends KafkaOperatorTestBase
private int totalBrokers = 0;
+ private String partition = null;
-
- @Parameterized.Parameters(name = "multi-cluster: {0}, multi-partition: {1}")
- public static Collection<Boolean[]> testScenario()
+ @Parameterized.Parameters(name = "multi-cluster: {0}, multi-partition: {1}, partition: {2}")
+ public static Collection<Object[]> testScenario()
{
- return Arrays.asList(new Boolean[][]{{true, false}, // multi cluster with single partition
- {true, true}, // multi cluster with multi partitions
- {false, true}, // single cluster with multi partitions
- {false, false}, // single cluster with single partitions
+ return Arrays.asList(new Object[][]{{true, false, "one_to_one"},// multi cluster with single partition
+ {true, false, "one_to_many"},
+ {true, true, "one_to_one"},// multi cluster with multi partitions
+ {true, true, "one_to_many"},
+ {false, true, "one_to_one"}, // single cluster with multi partitions
+ {false, true, "one_to_many"},
+ {false, false, "one_to_one"}, // single cluster with single partitions
+ {false, false, "one_to_many"}
});
}
- public KafkaInputOperatorTest(boolean hasMultiCluster, boolean hasMultiPartition)
+ public KafkaInputOperatorTest(boolean hasMultiCluster, boolean hasMultiPartition, String partition)
{
// This class want to initialize several kafka brokers for multiple partitions
this.hasMultiCluster = hasMultiCluster;
this.hasMultiPartition = hasMultiPartition;
int cluster = 1 + (hasMultiCluster ? 1 : 0);
totalBrokers = (1 + (hasMultiPartition ? 1 : 0)) * cluster;
-
+ this.partition = partition;
}
private static final org.slf4j.Logger logger = LoggerFactory.getLogger(KafkaInputOperatorTest.class);
@@ -167,6 +171,7 @@ public class KafkaInputOperatorTest extends KafkaOperatorTestBase
node.setTopics(TEST_TOPIC);
node.setInitialOffset(AbstractKafkaInputOperator.InitialOffset.EARLIEST.name());
node.setClusters(getClusterConfig());
+ node.setStrategy(partition);
// Create Test tuple collector
CollectorModule collector = dag.addOperator("TestMessageCollector", new CollectorModule());
[2/7] incubator-apex-malhar git commit: APEXMALHAR-1968 Update NOTICE
copyright year
Posted by th...@apache.org.
APEXMALHAR-1968 Update NOTICE copyright year
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/fa7402e7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/fa7402e7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/fa7402e7
Branch: refs/heads/release-3.3
Commit: fa7402e701dce519e0883d767a0074c6206b8ff4
Parents: cd183db
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Fri Jan 8 14:57:21 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:14:12 2016 -0800
----------------------------------------------------------------------
NOTICE | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/fa7402e7/NOTICE
----------------------------------------------------------------------
diff --git a/NOTICE b/NOTICE
index a511388..5a39f8e 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache Apex Malhar (incubating)
-Copyright (c) 2015 The Apache Software Foundation
+Copyright (c) 2015-2016 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
[6/7] incubator-apex-malhar git commit: APEXMALHAR-1983 #resolve
#comment split string in setter
Posted by th...@apache.org.
APEXMALHAR-1983 #resolve #comment split string in setter
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/c8af7e1e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/c8af7e1e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/c8af7e1e
Branch: refs/heads/release-3.3
Commit: c8af7e1ecc2a8164ef513b51c7f693198052a858
Parents: cbc1bbb
Author: Siyuan Hua <hs...@apache.org>
Authored: Fri Jan 29 21:22:30 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:24:21 2016 -0800
----------------------------------------------------------------------
.../malhar/kafka/AbstractKafkaInputOperator.java | 15 ++++++++++-----
.../apex/malhar/kafka/AbstractKafkaPartitioner.java | 4 +++-
.../apex/malhar/kafka/KafkaConsumerWrapper.java | 2 ++
.../org/apache/apex/malhar/kafka/KafkaMetrics.java | 2 ++
.../org/apache/apex/malhar/kafka/KafkaPartition.java | 3 +++
.../apex/malhar/kafka/OneToManyPartitioner.java | 2 ++
.../apex/malhar/kafka/OneToOnePartitioner.java | 2 ++
.../apache/apex/malhar/kafka/PartitionStrategy.java | 3 +++
8 files changed, 27 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
index 512f058..89104a3 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
@@ -34,12 +34,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
import com.datatorrent.api.AutoMetric;
import com.datatorrent.api.Context;
@@ -66,6 +69,7 @@ import com.datatorrent.api.StatsListener;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public abstract class AbstractKafkaInputOperator implements InputOperator, Operator.ActivationListener<Context.OperatorContext>, Operator.CheckpointListener, Partitioner<AbstractKafkaInputOperator>, StatsListener, OffsetCommitCallback
{
@@ -366,17 +370,18 @@ public abstract class AbstractKafkaInputOperator implements InputOperator, Opera
return Joiner.on(';').join(clusters);
}
- public void setTopics(String... topics)
+ public void setTopics(String topics)
{
- this.topics = topics;
+ this.topics = Iterables.toArray(Splitter.on(',').trimResults().omitEmptyStrings().split(topics), String.class);
}
/**
- * The topics the operator consumes
+ * The topics the operator consumes, separate by','
+ * Topic name can only contain ASCII alphanumerics, '.', '_' and '-'
*/
- public String[] getTopics()
+ public String getTopics()
{
- return topics;
+ return Joiner.on(", ").join(topics);
}
public void setStrategy(String policy)
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
index bf36064..7bb8585 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
@@ -32,6 +32,7 @@ import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
@@ -53,6 +54,7 @@ import com.datatorrent.lib.util.KryoCloneUtils;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public abstract class AbstractKafkaPartitioner implements Partitioner<AbstractKafkaInputOperator>, StatsListener
{
@@ -264,7 +266,7 @@ public abstract class AbstractKafkaPartitioner implements Partitioner<AbstractKa
{
return "PartitionMeta{" +
"cluster='" + cluster + '\'' +
- ", topicPartition=" + topicPartition +
+ ", topicPartition=" + getTopicPartition() +
'}';
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaConsumerWrapper.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaConsumerWrapper.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaConsumerWrapper.java
index cac2ad2..7a1211a 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaConsumerWrapper.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaConsumerWrapper.java
@@ -37,6 +37,7 @@ import org.slf4j.LoggerFactory;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -63,6 +64,7 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public class KafkaConsumerWrapper implements Closeable
{
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaMetrics.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaMetrics.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaMetrics.java
index 12a375d..75449a1 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaMetrics.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaMetrics.java
@@ -23,6 +23,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
@@ -33,6 +34,7 @@ import com.datatorrent.api.AutoMetric;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public class KafkaMetrics implements Serializable
{
private KafkaConsumerStats[] stats;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaPartition.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaPartition.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaPartition.java
index 4a4ebf3..1646ffe 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaPartition.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/KafkaPartition.java
@@ -21,9 +21,12 @@ package org.apache.apex.malhar.kafka;
import java.io.Serializable;
+import org.apache.hadoop.classification.InterfaceStability;
+
/**
* @since 2.1.0
*/
+@InterfaceStability.Evolving
public class KafkaPartition implements Serializable
{
protected static final String DEFAULT_CLUSTERID = "com.datatorrent.contrib.kafka.defaultcluster";
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
index bcd3073..3b4d3f3 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToManyPartitioner.java
@@ -24,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.common.PartitionInfo;
/**
@@ -32,6 +33,7 @@ import org.apache.kafka.common.PartitionInfo;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public class OneToManyPartitioner extends AbstractKafkaPartitioner
{
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToOnePartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToOnePartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToOnePartitioner.java
index b787932..570bdea 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToOnePartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/OneToOnePartitioner.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.hadoop.classification.InterfaceStability;
import org.apache.kafka.common.PartitionInfo;
import com.google.common.collect.Sets;
@@ -33,6 +34,7 @@ import com.google.common.collect.Sets;
*
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public class OneToOnePartitioner extends AbstractKafkaPartitioner
{
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/c8af7e1e/kafka/src/main/java/org/apache/apex/malhar/kafka/PartitionStrategy.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/PartitionStrategy.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/PartitionStrategy.java
index aaa35ec..7c142c5 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/PartitionStrategy.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/PartitionStrategy.java
@@ -18,9 +18,12 @@
*/
package org.apache.apex.malhar.kafka;
+import org.apache.hadoop.classification.InterfaceStability;
+
/**
* @since 3.3.0
*/
+@InterfaceStability.Evolving
public enum PartitionStrategy
{
/**
[5/7] incubator-apex-malhar git commit: APEXMALHAR-1984 #resolve
#comment Create util function to clone operator object by kryo
Posted by th...@apache.org.
APEXMALHAR-1984 #resolve #comment Create util function to clone operator object by kryo
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/cbc1bbbf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/cbc1bbbf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/cbc1bbbf
Branch: refs/heads/release-3.3
Commit: cbc1bbbf20e60ee2d25d1a666b1f7a3674e3688a
Parents: ca1c7e6
Author: Siyuan Hua <hs...@apache.org>
Authored: Thu Jan 28 15:30:04 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:18:39 2016 -0800
----------------------------------------------------------------------
.../AbstractCouchBaseInputOperator.java | 18 +-
.../kafka/AbstractKafkaInputOperator.java | 16 +-
.../kinesis/AbstractKinesisInputOperator.java | 16 +-
.../contrib/kafka/SimpleKakfaConsumerTest.java | 5 +-
.../util/FieldValueSerializableGenerator.java | 1 +
.../malhar/kafka/AbstractKafkaPartitioner.java | 14 +-
.../lib/io/block/AbstractBlockReader.java | 20 +--
.../lib/io/fs/AbstractFileInputOperator.java | 15 +-
.../datatorrent/lib/util/KryoCloneUtils.java | 171 +++++++++++++++++++
.../datastructs/DimensionalTableTest.java | 7 +-
.../appdata/schemas/ResultFormatterTest.java | 6 +-
.../schemas/SchemaRegistryMultipleTest.java | 5 +-
.../snapshot/AppDataSnapshotServerMapTest.java | 5 +-
.../lib/util/KryoCloneUtilsTest.java | 131 ++++++++++++++
.../com/datatorrent/lib/util/PojoUtilsTest.java | 3 +-
.../com/datatorrent/lib/util/TestUtils.java | 20 ---
16 files changed, 335 insertions(+), 118 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/contrib/src/main/java/com/datatorrent/contrib/couchbase/AbstractCouchBaseInputOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/couchbase/AbstractCouchBaseInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/couchbase/AbstractCouchBaseInputOperator.java
index 60938ab..1cd4eb5 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/couchbase/AbstractCouchBaseInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/couchbase/AbstractCouchBaseInputOperator.java
@@ -26,23 +26,19 @@ import java.util.concurrent.TimeUnit;
import com.couchbase.client.CouchbaseClient;
import com.couchbase.client.vbucket.config.Config;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.commons.io.output.ByteArrayOutputStream;
-
import com.datatorrent.lib.db.AbstractStoreInputOperator;
import com.datatorrent.api.Context;
import com.datatorrent.api.DefaultPartition;
import com.datatorrent.api.Partitioner;
+import com.datatorrent.lib.util.KryoCloneUtils;
import com.datatorrent.netlet.util.DTThrowable;
/**
@@ -149,19 +145,13 @@ public abstract class AbstractCouchBaseInputOperator<T> extends AbstractStoreInp
int numPartitions = conf.getServers().size();
List<String> list = conf.getServers();
Collection<Partition<AbstractCouchBaseInputOperator<T>>> newPartitions = Lists.newArrayListWithExpectedSize(numPartitions);
- Kryo kryo = new Kryo();
+ KryoCloneUtils<AbstractCouchBaseInputOperator<T>> cloneUtils = KryoCloneUtils.createCloneUtils(this);
for (int i = 0; i < numPartitions; i++) {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output output = new Output(bos);
- kryo.writeObject(output, this);
- output.close();
- Input lInput = new Input(bos.toByteArray());
- @SuppressWarnings("unchecked")
- AbstractCouchBaseInputOperator<T> oper = kryo.readObject(lInput, this.getClass());
+ AbstractCouchBaseInputOperator<T> oper = cloneUtils.getClone();
oper.setServerIndex(i);
oper.setServerURIString(list.get(i));
logger.debug("oper {} urlstring is {}", i, oper.getServerURIString());
- newPartitions.add(new DefaultPartition<AbstractCouchBaseInputOperator<T>>(oper));
+ newPartitions.add(new DefaultPartition<>(oper));
}
return newPartitions;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java
index 671a76f..b166b9e 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/kafka/AbstractKafkaInputOperator.java
@@ -29,9 +29,8 @@ import com.datatorrent.api.StatsListener;
import com.datatorrent.api.annotation.OperatorAnnotation;
import com.datatorrent.api.annotation.Stateless;
import com.datatorrent.lib.io.IdempotentStorageManager;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
+import com.datatorrent.lib.util.KryoCloneUtils;
+
import com.google.common.base.Joiner;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterators;
@@ -54,7 +53,6 @@ import org.slf4j.LoggerFactory;
import javax.validation.Valid;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Array;
import java.util.ArrayList;
@@ -640,14 +638,8 @@ public abstract class AbstractKafkaInputOperator<K extends KafkaConsumer> implem
// Create a new partition with the partition Ids and initial offset positions
protected Partitioner.Partition<AbstractKafkaInputOperator<K>> createPartition(Set<KafkaPartition> pIds, Map<KafkaPartition, Long> initOffsets, Collection<IdempotentStorageManager> newManagers)
{
- Kryo kryo = new Kryo();
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output output = new Output(bos);
- kryo.writeObject(output, this);
- output.close();
- Input lInput = new Input(bos.toByteArray());
- @SuppressWarnings("unchecked")
- Partitioner.Partition<AbstractKafkaInputOperator<K>> p = new DefaultPartition<AbstractKafkaInputOperator<K>>(kryo.readObject(lInput, this.getClass()));
+
+ Partitioner.Partition<AbstractKafkaInputOperator<K>> p = new DefaultPartition<>(KryoCloneUtils.cloneObject(this));
if (p.getPartitionedInstance().getConsumer() instanceof SimpleKafkaConsumer) {
p.getPartitionedInstance().getConsumer().resetPartitionsAndOffset(pIds, initOffsets);
if (initOffsets != null) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/contrib/src/main/java/com/datatorrent/contrib/kinesis/AbstractKinesisInputOperator.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/com/datatorrent/contrib/kinesis/AbstractKinesisInputOperator.java b/contrib/src/main/java/com/datatorrent/contrib/kinesis/AbstractKinesisInputOperator.java
index e7186b5..6306b04 100644
--- a/contrib/src/main/java/com/datatorrent/contrib/kinesis/AbstractKinesisInputOperator.java
+++ b/contrib/src/main/java/com/datatorrent/contrib/kinesis/AbstractKinesisInputOperator.java
@@ -26,11 +26,9 @@ import com.datatorrent.api.*;
import com.datatorrent.api.Operator.ActivationListener;
import com.datatorrent.common.util.Pair;
import com.datatorrent.lib.io.IdempotentStorageManager;
+import com.datatorrent.lib.util.KryoCloneUtils;
+
import com.esotericsoftware.kryo.DefaultSerializer;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.NotNull;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
import com.google.common.collect.Sets;
@@ -40,8 +38,8 @@ import org.slf4j.LoggerFactory;
import javax.validation.Valid;
import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.reflect.Array;
import java.util.*;
@@ -374,13 +372,7 @@ public abstract class AbstractKinesisInputOperator <T> implements InputOperator,
private
Partition<AbstractKinesisInputOperator> createPartition(Set<String> shardIds, Map<String, String> initShardPos, Collection<IdempotentStorageManager> newManagers)
{
- Kryo kryo = new Kryo();
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output output = new Output(bos);
- kryo.writeObject(output, this);
- output.close();
- Input lInput = new Input(bos.toByteArray());
- Partition<AbstractKinesisInputOperator> p = new DefaultPartition<AbstractKinesisInputOperator>(kryo.readObject(lInput, this.getClass()));
+ Partition<AbstractKinesisInputOperator> p = new DefaultPartition<AbstractKinesisInputOperator>(KryoCloneUtils.cloneObject(this));
newManagers.add(p.getPartitionedInstance().idempotentStorageManager);
p.getPartitionedInstance().getConsumer().setShardIds(shardIds);
p.getPartitionedInstance().getConsumer().resetShardPositions(initShardPos);
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/contrib/src/test/java/com/datatorrent/contrib/kafka/SimpleKakfaConsumerTest.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/kafka/SimpleKakfaConsumerTest.java b/contrib/src/test/java/com/datatorrent/contrib/kafka/SimpleKakfaConsumerTest.java
index d2491e1..1368cc0 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/kafka/SimpleKakfaConsumerTest.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/kafka/SimpleKakfaConsumerTest.java
@@ -18,8 +18,7 @@
*/
package com.datatorrent.contrib.kafka;
-import com.datatorrent.lib.util.TestUtils;
-import com.esotericsoftware.kryo.Kryo;
+import com.datatorrent.lib.util.KryoCloneUtils;
import org.junit.Assert;
import org.junit.Test;
@@ -39,7 +38,7 @@ public class SimpleKakfaConsumerTest
kc.setTopic("test_topic");
kc.setClientId("test_clientid");
- SimpleKafkaConsumer kcClone = TestUtils.clone(new Kryo(), kc);
+ SimpleKafkaConsumer kcClone = KryoCloneUtils.cloneObject(kc);
Assert.assertEquals("Buffer size is " + bufferSize, bufferSize, kcClone.getBufferSize());
Assert.assertEquals("Cache size is " + cacheSize, cacheSize, kcClone.getCacheSize());
Assert.assertEquals("Clint id is same", kc.getClientId(), kcClone.getClientId());
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/contrib/src/test/java/com/datatorrent/contrib/util/FieldValueSerializableGenerator.java
----------------------------------------------------------------------
diff --git a/contrib/src/test/java/com/datatorrent/contrib/util/FieldValueSerializableGenerator.java b/contrib/src/test/java/com/datatorrent/contrib/util/FieldValueSerializableGenerator.java
index 5340b0e..2975c9c 100644
--- a/contrib/src/test/java/com/datatorrent/contrib/util/FieldValueSerializableGenerator.java
+++ b/contrib/src/test/java/com/datatorrent/contrib/util/FieldValueSerializableGenerator.java
@@ -123,6 +123,7 @@ public class FieldValueSerializableGenerator< T extends FieldInfo> extends Field
{
if( _kryo == null )
_kryo = new Kryo();
+ _kryo.setClassLoader(clazz.getClassLoader());
}
}
return _kryo;
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
index 53bbd2a..bf36064 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
@@ -18,7 +18,6 @@
*/
package org.apache.apex.malhar.kafka;
-import java.io.ByteArrayOutputStream;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -38,14 +37,12 @@ import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
import com.google.common.base.Joiner;
import com.datatorrent.api.DefaultPartition;
import com.datatorrent.api.Partitioner;
import com.datatorrent.api.StatsListener;
+import com.datatorrent.lib.util.KryoCloneUtils;
/**
* Abstract partitioner used to manage the partitions of kafka input operator.
@@ -161,14 +158,7 @@ public abstract class AbstractKafkaPartitioner implements Partitioner<AbstractKa
protected Partitioner.Partition<AbstractKafkaInputOperator> createPartition(Set<AbstractKafkaPartitioner.PartitionMeta> partitionAssignment)
{
- Kryo kryo = new Kryo();
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output output = new Output(bos);
- kryo.writeObject(output, prototypeOperator);
- output.close();
- Input lInput = new Input(bos.toByteArray());
- Partitioner.Partition<AbstractKafkaInputOperator> p = (Partitioner.Partition<AbstractKafkaInputOperator>)
- new DefaultPartition<>(kryo.readObject(lInput, prototypeOperator.getClass()));
+ Partitioner.Partition<AbstractKafkaInputOperator> p = new DefaultPartition<AbstractKafkaInputOperator>(KryoCloneUtils.cloneObject(prototypeOperator));
p.getPartitionedInstance().assign(partitionAssignment);
return p;
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/main/java/com/datatorrent/lib/io/block/AbstractBlockReader.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/com/datatorrent/lib/io/block/AbstractBlockReader.java b/library/src/main/java/com/datatorrent/lib/io/block/AbstractBlockReader.java
index afde623..a6d1bd9 100644
--- a/library/src/main/java/com/datatorrent/lib/io/block/AbstractBlockReader.java
+++ b/library/src/main/java/com/datatorrent/lib/io/block/AbstractBlockReader.java
@@ -18,7 +18,6 @@
*/
package com.datatorrent.lib.io.block;
-import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
@@ -36,9 +35,6 @@ import org.slf4j.LoggerFactory;
import org.apache.commons.lang.mutable.MutableLong;
import org.apache.hadoop.fs.PositionedReadable;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -52,6 +48,7 @@ import com.datatorrent.api.Stats;
import com.datatorrent.api.StatsListener;
import com.datatorrent.common.util.BaseOperator;
import com.datatorrent.lib.counters.BasicCounters;
+import com.datatorrent.lib.util.KryoCloneUtils;
/**
* AbstractBlockReader processes a block of data from a stream.<br/>
@@ -295,20 +292,9 @@ public abstract class AbstractBlockReader<R, B extends BlockMetadata, STREAM ext
partitionIterator.remove();
}
} else {
- //Add more partitions
- Kryo kryo = new Kryo();
+ KryoCloneUtils<AbstractBlockReader<R, B, STREAM>> cloneUtils = KryoCloneUtils.createCloneUtils(this);
while (morePartitionsToCreate-- > 0) {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output loutput = new Output(bos);
- kryo.writeObject(loutput, this);
- loutput.close();
- Input lInput = new Input(bos.toByteArray());
-
- @SuppressWarnings("unchecked")
- AbstractBlockReader<R, B, STREAM> blockReader = kryo.readObject(lInput, this.getClass());
-
- DefaultPartition<AbstractBlockReader<R, B, STREAM>> partition = new DefaultPartition<>(
- blockReader);
+ DefaultPartition<AbstractBlockReader<R, B, STREAM>> partition = new DefaultPartition<>(cloneUtils.getClone());
newPartitions.add(partition);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/main/java/com/datatorrent/lib/io/fs/AbstractFileInputOperator.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/com/datatorrent/lib/io/fs/AbstractFileInputOperator.java b/library/src/main/java/com/datatorrent/lib/io/fs/AbstractFileInputOperator.java
index 5067b07..0bcf956 100644
--- a/library/src/main/java/com/datatorrent/lib/io/fs/AbstractFileInputOperator.java
+++ b/library/src/main/java/com/datatorrent/lib/io/fs/AbstractFileInputOperator.java
@@ -25,9 +25,6 @@ import java.util.regex.Pattern;
import javax.validation.constraints.NotNull;
-import com.esotericsoftware.kryo.Kryo;
-import com.esotericsoftware.kryo.io.Input;
-import com.esotericsoftware.kryo.io.Output;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
@@ -45,6 +42,7 @@ import com.datatorrent.lib.io.IdempotentStorageManager;
import com.datatorrent.api.*;
import com.datatorrent.api.Context.CountersAggregator;
import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.lib.util.KryoCloneUtils;
/**
* This is the base implementation of a directory input operator, which scans a directory for files.
@@ -829,21 +827,14 @@ public abstract class AbstractFileInputOperator<T> implements InputOperator, Par
*/
List<DirectoryScanner> scanners = scanner.partition(totalCount, oldscanners);
- Kryo kryo = new Kryo();
Collection<Partition<AbstractFileInputOperator<T>>> newPartitions = Lists.newArrayListWithExpectedSize(totalCount);
Collection<IdempotentStorageManager> newManagers = Lists.newArrayListWithExpectedSize(totalCount);
+ KryoCloneUtils<AbstractFileInputOperator<T>> cloneUtils = KryoCloneUtils.createCloneUtils(this);
for (int i=0; i<scanners.size(); i++) {
- // Kryo.copy fails as it attempts to clone transient fields
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output loutput = new Output(bos);
- kryo.writeObject(loutput, this);
- loutput.close();
- Input lInput = new Input(bos.toByteArray());
@SuppressWarnings("unchecked")
- AbstractFileInputOperator<T> oper = kryo.readObject(lInput, this.getClass());
- lInput.close();
+ AbstractFileInputOperator<T> oper = cloneUtils.getClone();
DirectoryScanner scn = scanners.get(i);
oper.setScanner(scn);
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/main/java/com/datatorrent/lib/util/KryoCloneUtils.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/com/datatorrent/lib/util/KryoCloneUtils.java b/library/src/main/java/com/datatorrent/lib/util/KryoCloneUtils.java
new file mode 100644
index 0000000..fbe6ce4
--- /dev/null
+++ b/library/src/main/java/com/datatorrent/lib/util/KryoCloneUtils.java
@@ -0,0 +1,171 @@
+/**
+ * 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 com.datatorrent.lib.util;
+
+import java.io.ByteArrayOutputStream;
+import java.lang.reflect.Array;
+
+import org.apache.commons.io.IOUtils;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Input;
+import com.esotericsoftware.kryo.io.Output;
+
+/**
+ *
+ * A Kryo Clone Util class that clone object by using Kryo serializer and deserializer
+ * The class has static method that are can be used directly to clone one object
+ * Or it can be used as util instance to clone as many objects as you need from the one source object
+ *
+ * @since 3.4.0
+ */
+public class KryoCloneUtils<T>
+{
+
+ /**
+ * Reusable Kryo object as deserializer
+ */
+ private final Kryo kryo;
+
+ /**
+ * Reusable binary data for object that would be deserialized from
+ */
+ private final byte[] bin;
+
+ /**
+ * The class of the object
+ */
+ private final Class<T> clazz;
+
+ @SuppressWarnings("unchecked")
+ private KryoCloneUtils(Kryo kryo, T t)
+ {
+ this.kryo = kryo;
+ ByteArrayOutputStream bos = null;
+ Output output = null;
+ try {
+ bos = new ByteArrayOutputStream();
+ output = new Output(bos);
+ kryo.writeObject(output, t);
+ output.close();
+ bin = bos.toByteArray();
+ } finally {
+ IOUtils.closeQuietly(output);
+ IOUtils.closeQuietly(bos);
+ }
+ clazz = (Class<T>)t.getClass();
+ kryo.setClassLoader(clazz.getClassLoader());
+ }
+
+ /**
+ * Clone from the binary data of the source object
+ * @return T
+ */
+ public T getClone()
+ {
+ try (Input input = new Input(bin)) {
+ return kryo.readObject(input, clazz);
+ }
+ }
+
+ /**
+ * Clone array of objects from source object
+ * @param num size of the return array
+ * @return array of T
+ */
+ @SuppressWarnings("unchecked")
+ public T[] getClones(int num)
+ {
+ T[] ts = (T[])Array.newInstance(clazz, num);
+ try (Input input = new Input(bin)) {
+ for (int i = 0; i < ts.length; i++) {
+ input.rewind();
+ ts[i] = kryo.readObject(input, clazz);
+ }
+ }
+ return ts;
+ }
+
+
+
+ /**
+ * Clone object by serializing and deserializing using Kryo.
+ * Note this is different from using {@link Kryo#copy(Object)}, which will attempt to also clone transient fields.
+ *
+ * @param kryo kryo object used to clone objects
+ * @param src src object that copy from
+ * @return
+ */
+ @SuppressWarnings("unchecked")
+ public static <SRC> SRC cloneObject(Kryo kryo, SRC src)
+ {
+ kryo.setClassLoader(src.getClass().getClassLoader());
+ ByteArrayOutputStream bos = null;
+ Output output;
+ Input input = null;
+ try {
+ bos = new ByteArrayOutputStream();
+ output = new Output(bos);
+ kryo.writeObject(output, src);
+ output.close();
+ input = new Input(bos.toByteArray());
+ return (SRC)kryo.readObject(input, src.getClass());
+ } finally {
+ IOUtils.closeQuietly(input);
+ IOUtils.closeQuietly(bos);
+ }
+ }
+
+
+ /**
+ * Clone object by serializing and deserializing using default Kryo.
+ * Note this is different from using {@link Kryo#copy(Object)}, which will attempt to also clone transient fields.
+ *
+ * @param src src object that copy from
+ * @return
+ */
+ public static <SRC> SRC cloneObject(SRC src)
+ {
+ return cloneObject(new Kryo(), src);
+ }
+
+ /**
+ * Factory function to return CloneUtils object
+ * @param template
+ * @param <SRC>
+ * @return
+ */
+ public static <SRC> KryoCloneUtils<SRC> createCloneUtils(SRC template)
+ {
+ return createCloneUtils(new Kryo(), template);
+ }
+
+ /**
+ * Factory function to return CloneUtils object with customized Kryo
+ * @param kryo
+ * @param template
+ * @param <SRC>
+ * @return
+ */
+ public static <SRC> KryoCloneUtils<SRC> createCloneUtils(Kryo kryo, SRC template)
+ {
+ return new KryoCloneUtils<>(kryo, template);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/appdata/datastructs/DimensionalTableTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/appdata/datastructs/DimensionalTableTest.java b/library/src/test/java/com/datatorrent/lib/appdata/datastructs/DimensionalTableTest.java
index bd2c660..db74007 100644
--- a/library/src/test/java/com/datatorrent/lib/appdata/datastructs/DimensionalTableTest.java
+++ b/library/src/test/java/com/datatorrent/lib/appdata/datastructs/DimensionalTableTest.java
@@ -21,7 +21,6 @@ package com.datatorrent.lib.appdata.datastructs;
import java.util.Map;
import java.util.Set;
-import com.esotericsoftware.kryo.Kryo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -29,7 +28,7 @@ import com.google.common.collect.Sets;
import org.junit.Assert;
import org.junit.Test;
-import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.lib.util.KryoCloneUtils;
public class DimensionalTableTest
{
@@ -38,7 +37,7 @@ public class DimensionalTableTest
{
DimensionalTable<Integer> table = createTestTable();
- TestUtils.clone(new Kryo(), table);
+ KryoCloneUtils.cloneObject(table);
}
@Test
@@ -47,7 +46,7 @@ public class DimensionalTableTest
DimensionalTable<Integer> table = createTestTable();
int size = table.size();
- table = TestUtils.clone(new Kryo(), table);
+ table = KryoCloneUtils.cloneObject(table);
Assert.assertEquals(size, table.size());
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/appdata/schemas/ResultFormatterTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/appdata/schemas/ResultFormatterTest.java b/library/src/test/java/com/datatorrent/lib/appdata/schemas/ResultFormatterTest.java
index 1f1895b..5f2d924 100644
--- a/library/src/test/java/com/datatorrent/lib/appdata/schemas/ResultFormatterTest.java
+++ b/library/src/test/java/com/datatorrent/lib/appdata/schemas/ResultFormatterTest.java
@@ -18,19 +18,17 @@
*/
package com.datatorrent.lib.appdata.schemas;
-import com.esotericsoftware.kryo.Kryo;
-
import org.junit.Assert;
import org.junit.Test;
-import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.lib.util.KryoCloneUtils;
public class ResultFormatterTest
{
@Test
public void serializationTest() throws Exception
{
- TestUtils.clone(new Kryo(), new ResultFormatter());
+ KryoCloneUtils.cloneObject(new ResultFormatter());
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/appdata/schemas/SchemaRegistryMultipleTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/appdata/schemas/SchemaRegistryMultipleTest.java b/library/src/test/java/com/datatorrent/lib/appdata/schemas/SchemaRegistryMultipleTest.java
index 8d997d8..0513079 100644
--- a/library/src/test/java/com/datatorrent/lib/appdata/schemas/SchemaRegistryMultipleTest.java
+++ b/library/src/test/java/com/datatorrent/lib/appdata/schemas/SchemaRegistryMultipleTest.java
@@ -21,14 +21,13 @@ package com.datatorrent.lib.appdata.schemas;
import java.util.Collections;
import java.util.Map;
-import com.esotericsoftware.kryo.Kryo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.junit.Assert;
import org.junit.Test;
-import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.lib.util.KryoCloneUtils;
public class SchemaRegistryMultipleTest
{
@@ -50,7 +49,7 @@ public class SchemaRegistryMultipleTest
{
SchemaRegistryMultiple schemaRegistryMultiple = createSchemaRegistry();
- schemaRegistryMultiple = TestUtils.clone(new Kryo(), schemaRegistryMultiple);
+ schemaRegistryMultiple = KryoCloneUtils.cloneObject(schemaRegistryMultiple);
Assert.assertEquals(2, schemaRegistryMultiple.size());
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/appdata/snapshot/AppDataSnapshotServerMapTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/appdata/snapshot/AppDataSnapshotServerMapTest.java b/library/src/test/java/com/datatorrent/lib/appdata/snapshot/AppDataSnapshotServerMapTest.java
index e02e2c0..96b348c 100644
--- a/library/src/test/java/com/datatorrent/lib/appdata/snapshot/AppDataSnapshotServerMapTest.java
+++ b/library/src/test/java/com/datatorrent/lib/appdata/snapshot/AppDataSnapshotServerMapTest.java
@@ -21,7 +21,6 @@ package com.datatorrent.lib.appdata.snapshot;
import java.util.List;
import java.util.Map;
-import com.esotericsoftware.kryo.Kryo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -32,7 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datatorrent.lib.testbench.CollectorTestSink;
-import com.datatorrent.lib.util.TestUtils;
+import com.datatorrent.lib.util.KryoCloneUtils;
public class AppDataSnapshotServerMapTest
{
@@ -103,7 +102,7 @@ public class AppDataSnapshotServerMapTest
Assert.assertEquals("b", secondRow.getString("word"));
//Test serialization
- TestUtils.clone(new Kryo(), snapshotServer);
+ KryoCloneUtils.cloneObject(snapshotServer);
}
private static final Logger LOG = LoggerFactory.getLogger(AppDataSnapshotServerMapTest.class);
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/util/KryoCloneUtilsTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/util/KryoCloneUtilsTest.java b/library/src/test/java/com/datatorrent/lib/util/KryoCloneUtilsTest.java
new file mode 100644
index 0000000..5eaea2e
--- /dev/null
+++ b/library/src/test/java/com/datatorrent/lib/util/KryoCloneUtilsTest.java
@@ -0,0 +1,131 @@
+/**
+ * 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 com.datatorrent.lib.util;
+
+import java.util.Objects;
+
+import org.junit.Test;
+
+import org.apache.commons.lang.math.RandomUtils;
+import org.apache.commons.lang3.RandomStringUtils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+/**
+ * Unit test for KryoCloneUtils
+ */
+public class KryoCloneUtilsTest
+{
+
+ @Test
+ public void testGetClone() throws Exception
+ {
+ TestEntity from = getTestEntity(5);
+ KryoCloneUtils<TestEntity> cloneUtils = KryoCloneUtils.createCloneUtils(from);
+ TestEntity to = cloneUtils.getClone();
+ assertFalse(from == to);
+ assertEquals(from, to);
+ assertFalse(from.transientProp.equals(to.transientProp));
+ }
+
+ @Test
+ public void testGetClones() throws Exception
+ {
+ TestEntity from = getTestEntity(5);
+ KryoCloneUtils<TestEntity> cloneUtils = KryoCloneUtils.createCloneUtils(from);
+ TestEntity[] to = cloneUtils.getClones(10);
+ for (TestEntity te : to) {
+ assertFalse(te == from);
+ assertEquals(from, te);
+ assertFalse(from.transientProp.equals(te.transientProp));
+ }
+ }
+
+ @Test
+ public void testCloneObject() throws Exception
+ {
+ TestEntity from = getTestEntity(5);
+ TestEntity to = KryoCloneUtils.cloneObject(from);
+ assertFalse(from == to);
+ assertEquals(from, to);
+ assertFalse(from.transientProp.equals(to.transientProp));
+ }
+
+ private TestEntity getTestEntity(int depth) {
+ TestEntity returnVal = null;
+ TestEntity curr = null;
+ while (depth-- > 0) {
+ if (curr == null) {
+ curr = returnVal = new TestEntity();
+ } else {
+ curr.nestedProp = new TestEntity();
+ curr = curr.nestedProp;
+ }
+ }
+ return returnVal;
+ }
+
+
+ static class TestEntity {
+
+ String strProp = RandomStringUtils.random(10);
+
+ int intProp = RandomUtils.nextInt(1000);
+
+ // transient should be skipped
+ transient String transientProp = RandomStringUtils.random(20);
+
+ // deep clone should be supported
+ TestEntity nestedProp = null;
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ TestEntity that = (TestEntity)o;
+ return Objects.equals(intProp, that.intProp) &&
+ Objects.equals(strProp, that.strProp) &&
+ Objects.equals(nestedProp, that.nestedProp);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(strProp, intProp, nestedProp);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "TestEntity{" +
+ "strProp='" + strProp + '\'' +
+ ", intProp=" + intProp +
+ ", transientProp='" + transientProp + '\'' +
+ ", nestedProp=" + nestedProp +
+ '}';
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/util/PojoUtilsTest.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/util/PojoUtilsTest.java b/library/src/test/java/com/datatorrent/lib/util/PojoUtilsTest.java
index 3114800..5b1b3f1 100644
--- a/library/src/test/java/com/datatorrent/lib/util/PojoUtilsTest.java
+++ b/library/src/test/java/com/datatorrent/lib/util/PojoUtilsTest.java
@@ -61,7 +61,6 @@ import com.datatorrent.lib.util.PojoUtils.SetterInt;
import com.datatorrent.lib.util.PojoUtils.SetterLong;
import com.datatorrent.lib.util.PojoUtils.SetterShort;
-import com.esotericsoftware.kryo.Kryo;
public class PojoUtilsTest
@@ -113,7 +112,7 @@ public class PojoUtilsTest
public void testSerialization() throws Exception
{
GetterBoolean<Object> getBoolean = createGetterBoolean(fqcn, "innerObj.boolVal");
- TestUtils.clone(new Kryo(), getBoolean);
+ KryoCloneUtils.cloneObject(getBoolean);
}
@Test
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/cbc1bbbf/library/src/test/java/com/datatorrent/lib/util/TestUtils.java
----------------------------------------------------------------------
diff --git a/library/src/test/java/com/datatorrent/lib/util/TestUtils.java b/library/src/test/java/com/datatorrent/lib/util/TestUtils.java
index 2932089..37d55e7 100644
--- a/library/src/test/java/com/datatorrent/lib/util/TestUtils.java
+++ b/library/src/test/java/com/datatorrent/lib/util/TestUtils.java
@@ -51,26 +51,6 @@ public class TestUtils
}
}
- /**
- * Clone object by serializing and deserializing using Kryo.
- * Note this is different from using {@link Kryo#copy(Object)}, which will attempt to also clone transient fields.
- * @param kryo
- * @param src
- * @return
- * @throws IOException
- */
- public static <T> T clone(Kryo kryo, T src) throws IOException
- {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- Output output = new Output(bos);
- kryo.writeObject(output, src);
- output.close();
- Input input = new Input(bos.toByteArray());
- @SuppressWarnings("unchecked")
- Class<T> clazz = (Class<T>)src.getClass();
- return kryo.readObject(input, clazz);
- }
-
@SuppressWarnings({ "unchecked", "rawtypes" })
public static <S extends Sink, T> S setSink(OutputPort<T> port, S sink)
{
[7/7] incubator-apex-malhar git commit: APEXMALHAR-1986 Change base
for semantic version check to 3.3 release.
Posted by th...@apache.org.
APEXMALHAR-1986 Change base for semantic version check to 3.3 release.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/040f8f77
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/040f8f77
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/040f8f77
Branch: refs/heads/release-3.3
Commit: 040f8f77762ecbc1d07b7f4201ce3d83fb1600e4
Parents: c8af7e1
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Thu Jan 28 17:51:50 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:26:51 2016 -0800
----------------------------------------------------------------------
pom.xml | 12 +++++++-----
1 file changed, 7 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/040f8f77/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 2c3f26a..3945d7f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -108,7 +108,7 @@
<dependency>
<groupId>org.apache.apex</groupId>
<artifactId>${project.artifactId}</artifactId>
- <version>3.2.0-incubating</version>
+ <version>3.3.0-incubating</version>
</dependency>
</oldVersion>
<newVersion>
@@ -125,10 +125,12 @@
<includeSynthetic>false</includeSynthetic>
<ignoreMissingClasses>true</ignoreMissingClasses>
<excludes>
- <exclude>com.datatorrent.contrib.converter.Converter</exclude>
- <exclude>com.datatorrent.contrib.schema</exclude>
- <exclude>org.apache.hadoop.io.file.tfile</exclude>
- <exclude>*POJO*</exclude>
+ <exclude>@org.apache.hadoop.classification.InterfaceStability.Evolving</exclude>
+ <exclude>@org.apache.hadoop.classification.InterfaceStability.Unstable</exclude>
+ <exclude>com.datatorrent.lib.parser.JsonParser</exclude>
+ <exclude>com.datatorrent.lib.parser.Parser</exclude>
+ <exclude>com.datatorrent.lib.parser.XmlParser</exclude>
+ <exclude>com.datatorrent.contrib.parser.CsvParser</exclude>
</excludes>
</parameter>
<skip>${semver.plugin.skip}</skip>
[4/7] incubator-apex-malhar git commit: APEXMALHAR-1973 #comment
disable committing offsets for latest|earliest and store the offset for next
message
Posted by th...@apache.org.
APEXMALHAR-1973 #comment disable committing offsets for latest|earliest and store the offset for next message
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/ca1c7e60
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/ca1c7e60
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/ca1c7e60
Branch: refs/heads/release-3.3
Commit: ca1c7e60f91779d7c19ecc70baa997d7ae2a2c47
Parents: be71996
Author: Siyuan Hua <hs...@apache.org>
Authored: Wed Jan 13 17:07:18 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:17:42 2016 -0800
----------------------------------------------------------------------
.../apache/apex/malhar/kafka/AbstractKafkaInputOperator.java | 4 +++-
.../apache/apex/malhar/kafka/AbstractKafkaPartitioner.java | 7 ++++---
2 files changed, 7 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/ca1c7e60/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
index fbff2e7..512f058 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaInputOperator.java
@@ -179,6 +179,8 @@ public abstract class AbstractKafkaInputOperator implements InputOperator, Opera
@Override
public void committed(long windowId)
{
+ if (initialOffset == InitialOffset.LATEST || initialOffset == InitialOffset.EARLIEST)
+ return;
//ask kafka consumer wrapper to store the committed offsets
for (Iterator<Pair<Long, Map<AbstractKafkaPartitioner.PartitionMeta, Long>>> iter = offsetHistory.iterator(); iter.hasNext(); ) {
Pair<Long, Map<AbstractKafkaPartitioner.PartitionMeta, Long>> item = iter.next();
@@ -204,7 +206,7 @@ public abstract class AbstractKafkaInputOperator implements InputOperator, Opera
emitTuple(tuple.getLeft(), msg);
AbstractKafkaPartitioner.PartitionMeta pm = new AbstractKafkaPartitioner.PartitionMeta(tuple.getLeft(),
msg.topic(), msg.partition());
- offsetTrack.put(pm, msg.offset());
+ offsetTrack.put(pm, msg.offset() + 1);
}
emitCount += count;
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/ca1c7e60/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
index c708145..53bbd2a 100644
--- a/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
+++ b/kafka/src/main/java/org/apache/apex/malhar/kafka/AbstractKafkaPartitioner.java
@@ -258,14 +258,15 @@ public abstract class AbstractKafkaPartitioner implements Partitioner<AbstractKa
return false;
}
PartitionMeta that = (PartitionMeta)o;
- return Objects.equals(cluster, that.cluster) &&
- Objects.equals(getTopicPartition(), that.getTopicPartition());
+ return Objects.equals(partitionId, that.partitionId) &&
+ Objects.equals(cluster, that.cluster) &&
+ Objects.equals(topic, that.topic);
}
@Override
public int hashCode()
{
- return Objects.hash(cluster, getTopicPartition());
+ return Objects.hash(cluster, topic, partitionId);
}
@Override
[3/7] incubator-apex-malhar git commit: Skip deploy (modules that
should not go to central) independent from -Pall-modules.
Posted by th...@apache.org.
Skip deploy (modules that should not go to central) independent from -Pall-modules.
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/be71996a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/be71996a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/be71996a
Branch: refs/heads/release-3.3
Commit: be71996a07bed63140c58dc4635558af43985870
Parents: fa7402e
Author: Thomas Weise <th...@datatorrent.com>
Authored: Wed Jan 13 10:10:29 2016 -0800
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Sat Feb 20 00:16:38 2016 -0800
----------------------------------------------------------------------
apps/logstream/pom.xml | 2 +-
benchmark/pom.xml | 2 +-
library/pom.xml | 1 -
samples/pom.xml | 4 ++++
4 files changed, 6 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/be71996a/apps/logstream/pom.xml
----------------------------------------------------------------------
diff --git a/apps/logstream/pom.xml b/apps/logstream/pom.xml
index 3e79d75..2eb8f47 100644
--- a/apps/logstream/pom.xml
+++ b/apps/logstream/pom.xml
@@ -32,7 +32,7 @@
<packaging>jar</packaging>
<properties>
- <maven.deploy.skip>false</maven.deploy.skip>
+ <maven.deploy.skip>true</maven.deploy.skip>
<skipTests>false</skipTests>
<semver.plugin.skip>true</semver.plugin.skip>
</properties>
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/be71996a/benchmark/pom.xml
----------------------------------------------------------------------
diff --git a/benchmark/pom.xml b/benchmark/pom.xml
index 4fde9cc..318b865 100644
--- a/benchmark/pom.xml
+++ b/benchmark/pom.xml
@@ -37,7 +37,7 @@
<properties>
<datatorrent.apppackage.classpath>lib/*.jar</datatorrent.apppackage.classpath>
- <maven.deploy.skip>false</maven.deploy.skip>
+ <maven.deploy.skip>true</maven.deploy.skip>
<skipTests>true</skipTests>
<semver.plugin.skip>true</semver.plugin.skip>
</properties>
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/be71996a/library/pom.xml
----------------------------------------------------------------------
diff --git a/library/pom.xml b/library/pom.xml
index 2957175..7a568ed 100644
--- a/library/pom.xml
+++ b/library/pom.xml
@@ -35,7 +35,6 @@
<name>Apache Apex Malhar (incubating) Library</name>
<properties>
- <maven.deploy.skip>false</maven.deploy.skip>
</properties>
<build>
http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/be71996a/samples/pom.xml
----------------------------------------------------------------------
diff --git a/samples/pom.xml b/samples/pom.xml
index 8edfe5e..6e6ff9d 100644
--- a/samples/pom.xml
+++ b/samples/pom.xml
@@ -34,6 +34,10 @@
<name>Apache Apex Malhar (incubating) Samples</name>
+ <properties>
+ <maven.deploy.skip>true</maven.deploy.skip>
+ </properties>
+
<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>