You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/03/19 20:59:00 UTC
[2/3] flink git commit: [FLINK-1752][streaming] Add KafkaITCase and
various bugfixes
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/EncoderWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/EncoderWrapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/EncoderWrapper.java
deleted file mode 100644
index 52ea319..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/EncoderWrapper.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.config;
-
-import org.apache.flink.streaming.connectors.util.SerializationSchema;
-
-import kafka.serializer.Encoder;
-import kafka.utils.VerifiableProperties;
-
-/**
- * Wraps an arbitrary SerializationScheme to use as a Kafka Encoder.
- *
- * @param <T>
- * Type to serialize
- */
-public class EncoderWrapper<T> extends KafkaConfigWrapper<SerializationSchema<T, byte[]>> implements Encoder<T> {
-
- public EncoderWrapper(SerializationSchema<T, byte[]> wrapped) {
- super(wrapped);
- }
-
- public EncoderWrapper(VerifiableProperties properties) {
- super(properties);
- }
-
- @Override
- public byte[] toBytes(T element) {
- return wrapped.serialize(element);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/KafkaConfigWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/KafkaConfigWrapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/KafkaConfigWrapper.java
deleted file mode 100644
index af5fb63..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/KafkaConfigWrapper.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.config;
-
-import java.io.Serializable;
-import java.util.Properties;
-
-import kafka.utils.VerifiableProperties;
-
-/**
- * Wraps an arbitrary Serializable (e.g. a Partitioner) object to use in Kafka, to shade the properties.
- *
- * @param <T>
- * Type of the object to wrap
- */
-public abstract class KafkaConfigWrapper<T extends Serializable> {
-
- private StringSerializer<T> stringSerializer;
-
- protected T wrapped;
-
- public KafkaConfigWrapper(T wrapped) {
- this();
- this.wrapped = wrapped;
- }
-
- private KafkaConfigWrapper() {
- stringSerializer = new StringSerializer<T>();
- }
-
- public KafkaConfigWrapper(VerifiableProperties properties) {
- this();
- read(properties);
- }
-
- public void read(VerifiableProperties properties) {
- String stringWrapped = properties.getString(getClass().getCanonicalName());
- wrapped = stringSerializer.deserialize(stringWrapped);
- }
-
- public void write(Properties properties) {
- properties.put(getClass().getCanonicalName(), stringSerializer.serialize(wrapped));
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/PartitionerWrapper.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/PartitionerWrapper.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/PartitionerWrapper.java
deleted file mode 100644
index bfeed10..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/PartitionerWrapper.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.config;
-
-import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
-
-import kafka.producer.Partitioner;
-import kafka.utils.VerifiableProperties;
-
-/**
- * Wraps an arbitrary partitioner to use as a Kafka partitioner.
- *
- * @param <T>
- * Type to partition
- */
-public class PartitionerWrapper<T> extends KafkaConfigWrapper<KafkaPartitioner<T>> implements Partitioner {
-
- public PartitionerWrapper(KafkaPartitioner<T> wrapped) {
- super(wrapped);
- }
-
- public PartitionerWrapper(VerifiableProperties properties) {
- super(properties);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public int partition(Object key, int numPartitions) {
- return wrapped.partition((T) key, numPartitions);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StringSerializer.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StringSerializer.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StringSerializer.java
deleted file mode 100644
index d7ac198..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StringSerializer.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.config;
-
-import java.io.Serializable;
-
-import org.apache.commons.codec.binary.Base64;
-
-import org.apache.commons.lang3.SerializationUtils;
-
-/**
- * Serializer to serializer an arbitrary object to String.
- *
- * @param <T>
- * Type to serialize.
- */
-public class StringSerializer<T extends Serializable> {
-
- public String serialize(T element) {
- byte[] serialized = SerializationUtils.serialize(element);
- return Base64.encodeBase64String(serialized);
- }
-
- public T deserialize(String stringSerialized) {
- byte[] bytes = Base64.decodeBase64(stringSerialized);
- return SerializationUtils.deserialize(bytes);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaConstantPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaConstantPartitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaConstantPartitioner.java
index 4afd91b..661d0bd 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaConstantPartitioner.java
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaConstantPartitioner.java
@@ -17,7 +17,7 @@
package org.apache.flink.streaming.connectors.kafka.partitioner;
-public class KafkaConstantPartitioner<T> implements KafkaPartitioner<T> {
+public class KafkaConstantPartitioner implements SerializableKafkaPartitioner {
private static final long serialVersionUID = 1L;
private int partition;
@@ -27,8 +27,7 @@ public class KafkaConstantPartitioner<T> implements KafkaPartitioner<T> {
}
@Override
- public int partition(T value, int numberOfPartitions) {
+ public int partition(Object value, int numberOfPartitions) {
return partition;
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaDistributePartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaDistributePartitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaDistributePartitioner.java
deleted file mode 100644
index 480aa17..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaDistributePartitioner.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.partitioner;
-
-/**
- * Kafka partitioner that distributes the data equally by cycling through the output
- * channels.
- *
- * @param <T>
- * Type to partition.
- */
-public class KafkaDistributePartitioner<T> implements KafkaPartitioner<T> {
-
- private static final long serialVersionUID = 1L;
- int currentPartition;
-
- public KafkaDistributePartitioner() {
- currentPartition = 0;
- }
-
- @Override
- public int partition(T value, int numberOfPartitions) {
- return currentPartition++ % numberOfPartitions;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaPartitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaPartitioner.java
deleted file mode 100644
index f82ad7c..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/KafkaPartitioner.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka.partitioner;
-
-import java.io.Serializable;
-
-public interface KafkaPartitioner<T> extends Serializable {
-
- public int partition(T value, int numberOfPartitions);
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/SerializableKafkaPartitioner.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/SerializableKafkaPartitioner.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/SerializableKafkaPartitioner.java
new file mode 100644
index 0000000..77a774e
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/src/main/java/org/apache/flink/streaming/connectors/kafka/partitioner/SerializableKafkaPartitioner.java
@@ -0,0 +1,24 @@
+/*
+ * 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.flink.streaming.connectors.kafka.partitioner;
+
+import kafka.producer.Partitioner;
+import java.io.Serializable;
+
+public interface SerializableKafkaPartitioner extends Serializable, Partitioner {
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
new file mode 100644
index 0000000..134525d
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaITCase.java
@@ -0,0 +1,252 @@
+/*
+ * 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.flink.streaming.connectors.kafka;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.BitSet;
+import java.util.Properties;
+
+import org.apache.curator.test.TestingServer;
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.net.NetUtils;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.function.sink.SinkFunction;
+import org.apache.flink.streaming.api.function.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.api.KafkaSink;
+import org.apache.flink.streaming.connectors.kafka.api.simple.KafkaTopicUtils;
+import org.apache.flink.streaming.connectors.kafka.api.simple.PersistentKafkaSource;
+import org.apache.flink.streaming.connectors.kafka.api.simple.offset.Offset;
+import org.apache.flink.streaming.connectors.util.JavaDefaultStringSchema;
+import org.apache.flink.util.Collector;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import kafka.server.KafkaConfig;
+import kafka.server.KafkaServer;
+import kafka.utils.Time;
+
+/**
+ * Code in this test is based on the following GitHub repository:
+ * (as per commit bc6b2b2d5f6424d5f377aa6c0871e82a956462ef)
+ * <p/>
+ * https://github.com/sakserv/hadoop-mini-clusters (ASL licensed)
+ */
+
+public class KafkaITCase {
+
+ private static final Logger LOG = LoggerFactory.getLogger(KafkaITCase.class);
+
+ private final String TOPIC = "myTopic";
+
+ private int zkPort;
+ private int kafkaPort;
+ private String kafkaHost;
+
+ private String zookeeperConnectionString;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+ public File tmpZkDir;
+ public File tmpKafkaDir;
+
+ @Before
+ public void prepare() throws IOException {
+ tmpZkDir = tempFolder.newFolder();
+ tmpKafkaDir = tempFolder.newFolder();
+ kafkaHost = InetAddress.getLocalHost().getHostName();
+ zkPort = NetUtils.getAvailablePort();
+ kafkaPort = NetUtils.getAvailablePort();
+ zookeeperConnectionString = "localhost:" + zkPort;
+ }
+
+ @Test
+ public void test() {
+ LOG.info("Starting KafkaITCase.test()");
+ TestingServer zookeeper = null;
+ KafkaServer broker1 = null;
+ try {
+ zookeeper = getZookeeper();
+ broker1 = getKafkaServer(0);
+ LOG.info("ZK and KafkaServer started. Creating test topic:");
+ createTestTopic();
+
+ LOG.info("Starting Kafka Topology in Flink:");
+ startKafkaTopology();
+
+ LOG.info("Test succeeded.");
+ } catch (Throwable t) {
+ LOG.warn("Test failed with exception", t);
+ Assert.fail("Test failed with: " + t.getMessage());
+ } finally {
+ LOG.info("Shutting down all services");
+ if (broker1 != null) {
+ broker1.shutdown();
+ }
+ if (zookeeper != null) {
+ try {
+ zookeeper.stop();
+ } catch (IOException e) {
+ LOG.warn("ZK.stop() failed", e);
+ }
+ }
+ }
+
+ }
+
+ private void createTestTopic() {
+ KafkaTopicUtils kafkaTopicUtils = new KafkaTopicUtils(zookeeperConnectionString);
+ kafkaTopicUtils.createTopic(TOPIC, 1, 1);
+ }
+
+ private void startKafkaTopology() throws Exception {
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
+
+ // add consuming topology:
+ DataStreamSource<String> consuming = env.addSource(
+ new PersistentKafkaSource<String>(zookeeperConnectionString, TOPIC, new JavaDefaultStringSchema(), 5000, 100, Offset.FROM_BEGINNING));
+ consuming.addSink(new SinkFunction<String>() {
+ int elCnt = 0;
+ int start = -1;
+ BitSet validator = new BitSet(101);
+
+ @Override
+ public void invoke(String value) throws Exception {
+ LOG.info("Got " + value);
+ String[] sp = value.split("-");
+ int v = Integer.parseInt(sp[1]);
+ if (start == -1) {
+ start = v;
+ }
+ Assert.assertFalse("Received tuple twice", validator.get(v - start));
+ validator.set(v - start);
+ elCnt++;
+ if (elCnt == 100) {
+ // check if everything in the bitset is set to true
+ int nc;
+ if ((nc = validator.nextClearBit(0)) != 100) {
+ throw new RuntimeException("The bitset was not set to 1 on all elements. Next clear:" + nc + " Set: " + validator);
+ }
+ throw new SuccessException();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ }
+ });
+
+ // add producing topology
+ DataStream<String> stream = env.addSource(new SourceFunction<String>() {
+ boolean running = true;
+
+ @Override
+ public void run(Collector<String> collector) throws Exception {
+ LOG.info("Starting source.");
+ int cnt = 0;
+ while (running) {
+ collector.collect("kafka-" + cnt++);
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException ignored) {
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ LOG.info("Source got chancel()");
+ running = false;
+ }
+ });
+ stream.addSink(new KafkaSink<String>(zookeeperConnectionString, TOPIC, new JavaDefaultStringSchema()));
+
+ try {
+ env.setDegreeOfParallelism(1);
+ env.execute();
+ } catch (JobExecutionException good) {
+ Throwable t = good.getCause();
+ int limit = 0;
+ while (!(t instanceof SuccessException)) {
+ t = t.getCause();
+ if (limit++ == 20) {
+ throw good;
+ }
+ }
+ }
+ }
+
+
+ private TestingServer getZookeeper() throws Exception {
+ return new TestingServer(zkPort, tmpZkDir);
+ }
+
+ /**
+ * Copied from com.github.sakserv.minicluster.KafkaLocalBrokerIntegrationTest (ASL licensed)
+ */
+ private KafkaServer getKafkaServer(int brokerId) throws UnknownHostException {
+ Properties kafkaProperties = new Properties();
+ // properties have to be Strings
+ kafkaProperties.put("advertised.host.name", kafkaHost);
+ kafkaProperties.put("port", Integer.toString(kafkaPort));
+ kafkaProperties.put("broker.id", Integer.toString(brokerId));
+ kafkaProperties.put("log.dir", tmpKafkaDir.toString());
+ kafkaProperties.put("zookeeper.connect", zookeeperConnectionString);
+ KafkaConfig kafkaConfig = new KafkaConfig(kafkaProperties);
+
+ KafkaServer server = new KafkaServer(kafkaConfig, new LocalSystemTime());
+ server.startup();
+ return server;
+ }
+
+ public class LocalSystemTime implements Time {
+
+ @Override
+ public long milliseconds() {
+ return System.currentTimeMillis();
+ }
+
+ public long nanoseconds() {
+ return System.nanoTime();
+ }
+
+ @Override
+ public void sleep(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ LOG.warn("Interruption", e);
+ }
+ }
+
+ }
+
+ public static class SuccessException extends Exception {
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/StringSerializerTest.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/StringSerializerTest.java b/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/StringSerializerTest.java
deleted file mode 100644
index 0112927..0000000
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/test/java/org/apache/flink/streaming/connectors/kafka/StringSerializerTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.flink.streaming.connectors.kafka;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.Serializable;
-
-import org.apache.flink.streaming.connectors.kafka.config.StringSerializer;
-import org.junit.Test;
-
-public class StringSerializerTest {
-
- private static class MyClass implements Serializable {
-
- private static final long serialVersionUID = 1L;
- private int a;
- private String b;
-
- public MyClass(int a, String b) {
- this.a = a;
- this.b = b;
- }
-
- @Override
- public boolean equals(Object o) {
- if (o == null) {
- return false;
- } else {
- try {
- MyClass other = (MyClass) o;
- return a == other.a && b.equals(other.b);
- } catch (ClassCastException e) {
- return false;
- }
- }
- }
- }
-
- @Test
- public void test() {
-
- MyClass myObject = new MyClass(42, "test string");
-
- StringSerializer<MyClass> stringSerializer = new StringSerializer<MyClass>();
-
- String store = stringSerializer.serialize(myObject);
-
- MyClass myObjectDeserialized = stringSerializer.deserialize(store);
-
- assertEquals(myObject, myObjectDeserialized);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-connectors/src/test/resources/log4j-test.properties
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-connectors/src/test/resources/log4j-test.properties b/flink-staging/flink-streaming/flink-streaming-connectors/src/test/resources/log4j-test.properties
index 2fb9345..dc20726 100644
--- a/flink-staging/flink-streaming/flink-streaming-connectors/src/test/resources/log4j-test.properties
+++ b/flink-staging/flink-streaming/flink-streaming-connectors/src/test/resources/log4j-test.properties
@@ -16,4 +16,12 @@
# limitations under the License.
################################################################################
-log4j.rootLogger=OFF
\ No newline at end of file
+log4j.rootLogger=OFF, testlogger
+
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target = System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+
+# suppress the irrelevant (wrong) warnings from the netty channel handler
+log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, testlogger
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 53179d7..879a98c 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -257,7 +257,7 @@ public class DataStream<OUT> {
/**
* Creates a new {@link ConnectedDataStream} by connecting
- * {@link DataStream} outputs of (possible) different typea with each other.
+ * {@link DataStream} outputs of (possible) different types with each other.
* The DataStreams connected using this operator can be used with
* CoFunctions to apply joint transformations.
*
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 9cc6131..3a617dc 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -547,6 +548,7 @@ public abstract class StreamExecutionEnvironment {
boolean isParallel = function instanceof ParallelSourceFunction;
int dop = isParallel ? getDegreeOfParallelism() : 1;
+ ClosureCleaner.clean(function, true);
StreamInvokable<OUT, OUT> sourceInvokable = new SourceInvokable<OUT>(function);
DataStreamSource<OUT> returnStream = new DataStreamSource<OUT>(this, sourceName,
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/co/CoInvokable.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/co/CoInvokable.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/co/CoInvokable.java
index b036829..f727a32 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/co/CoInvokable.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/invokable/operator/co/CoInvokable.java
@@ -136,6 +136,7 @@ public abstract class CoInvokable<IN1, IN2, OUT> extends StreamInvokable<IN1, OU
LOG.error("Calling user function failed due to: {}",
StringUtils.stringifyException(e));
}
+ throw new RuntimeException(e);
}
}
@@ -147,6 +148,7 @@ public abstract class CoInvokable<IN1, IN2, OUT> extends StreamInvokable<IN1, OU
LOG.error("Calling user function failed due to: {}",
StringUtils.stringifyException(e));
}
+ throw new RuntimeException(e);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
index 3548712..926aac2 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/streamvertex/StreamVertex.java
@@ -109,7 +109,7 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
* there is user state it also checkpoints the state to the jobmanager.
*/
@Override
- public void confirmBarrier(long barrierID) {
+ public void confirmBarrier(long barrierID) throws IOException {
if (configuration.getStateMonitoring() && !states.isEmpty()) {
getEnvironment().getJobManager().tell(
@@ -271,10 +271,8 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
if (LOG.isDebugEnabled()) {
LOG.debug("Superstep " + id + " processed: " + StreamVertex.this);
}
- } catch (IOException e) {
- e.printStackTrace();
- } catch (InterruptedException e) {
- e.printStackTrace();
+ } catch (Exception e) {
+ throw new RuntimeException("Error while confirming barrier", e);
}
}
@@ -288,7 +286,7 @@ public class StreamVertex<IN, OUT> extends AbstractInvokable implements StreamTa
*/
@Override
public void injectState(StateHandle stateHandle) {
- this.states.putAll(stateHandle.getState());
+ this.states.putAll(stateHandle.getState(userClassLoader));
}
private class SuperstepEventListener implements EventListener<TaskEvent> {
http://git-wip-us.apache.org/repos/asf/flink/blob/c0917f23/flink-staging/flink-tachyon/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/flink-staging/flink-tachyon/src/test/resources/log4j.properties b/flink-staging/flink-tachyon/src/test/resources/log4j.properties
index 2c9a677..f533ba2 100644
--- a/flink-staging/flink-tachyon/src/test/resources/log4j.properties
+++ b/flink-staging/flink-tachyon/src/test/resources/log4j.properties
@@ -20,4 +20,12 @@
# Until the issue is resolved (see https://github.com/amplab/tachyon/pull/571)
# we provide a log4j.properties file ourselves.
-log4j.rootLogger=OFF
\ No newline at end of file
+log4j.rootLogger=OFF, testlogger
+
+log4j.appender.testlogger=org.apache.log4j.ConsoleAppender
+log4j.appender.testlogger.target = System.err
+log4j.appender.testlogger.layout=org.apache.log4j.PatternLayout
+log4j.appender.testlogger.layout.ConversionPattern=%-4r [%t] %-5p %c %x - %m%n
+
+# suppress the irrelevant (wrong) warnings from the netty channel handler
+log4j.logger.org.jboss.netty.channel.DefaultChannelPipeline=ERROR, testlogger
\ No newline at end of file