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/11/27 16:24:47 UTC
flink git commit: [FLINK-3054] Remove R (return) type variable from
SerializationSchema
Repository: flink
Updated Branches:
refs/heads/master 76913686c -> 09ea3dd0f
[FLINK-3054] Remove R (return) type variable from SerializationSchema
This closes #1406
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/09ea3dd0
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/09ea3dd0
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/09ea3dd0
Branch: refs/heads/master
Commit: 09ea3dd0f5ccbdeb381a840f4da65a99c11817a0
Parents: 7691368
Author: Robert Metzger <rm...@apache.org>
Authored: Tue Nov 24 17:58:00 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Fri Nov 27 16:24:03 2015 +0100
----------------------------------------------------------------------
docs/apis/streaming_guide.md | 2 +-
.../streaming/connectors/flume/FlumeSink.java | 4 +-
.../connectors/kafka/FlinkKafkaProducer.java | 6 +--
.../connectors/kafka/api/KafkaSink.java | 2 +-
.../connectors/kafka/KafkaConsumerTest.java | 4 +-
.../connectors/kafka/KafkaConsumerTestBase.java | 8 +--
.../connectors/kafka/KafkaProducerTest.java | 6 +--
.../kafka/testutils/DataGenerators.java | 4 +-
.../streaming/connectors/rabbitmq/RMQSink.java | 4 +-
.../connectors/rabbitmq/RMQTopology.java | 14 +-----
.../streaming/api/datastream/DataStream.java | 2 +-
.../api/functions/sink/SocketClientSink.java | 8 +--
.../serialization/JavaDefaultStringSchema.java | 47 ------------------
.../KeyedSerializationSchemaWrapper.java | 4 +-
.../streaming/util/serialization/RawSchema.java | 52 --------------------
.../util/serialization/SerializationSchema.java | 5 +-
.../util/serialization/SimpleStringSchema.java | 23 +++++++--
.../TypeInformationSerializationSchema.java | 2 +-
.../functions/sink/SocketClientSinkTest.java | 2 +-
.../outputformat/SocketOutputFormatITCase.java | 3 +-
.../streaming/util/SocketOutputTestBase.java | 28 +----------
.../flink/streaming/api/scala/DataStream.scala | 2 +-
.../api/scala/OutputFormatTestPrograms.scala | 5 +-
23 files changed, 60 insertions(+), 177 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/docs/apis/streaming_guide.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming_guide.md b/docs/apis/streaming_guide.md
index 7062a16..3bb597b 100644
--- a/docs/apis/streaming_guide.md
+++ b/docs/apis/streaming_guide.md
@@ -3463,7 +3463,7 @@ properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("zookeeper.connect", "localhost:2181");
properties.setProperty("group.id", "test");
stream = env
- .addSource(new KafkaSource[String]("topic", new SimpleStringSchema(), properties))
+ .addSource(new FlinkKafkaConsumer082[String]("topic", new SimpleStringSchema(), properties))
.print
{% endhighlight %}
</div>
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java b/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
index 50f5770..2dc043b 100644
--- a/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
+++ b/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java
@@ -39,9 +39,9 @@ public class FlumeSink<IN> extends RichSinkFunction<IN> {
boolean initDone = false;
String host;
int port;
- SerializationSchema<IN, byte[]> schema;
+ SerializationSchema<IN> schema;
- public FlumeSink(String host, int port, SerializationSchema<IN, byte[]> schema) {
+ public FlumeSink(String host, int port, SerializationSchema<IN> schema) {
this.host = host;
this.port = port;
this.schema = schema;
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
index 6fe66d8..a8d913b 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java
@@ -113,7 +113,7 @@ public class FlinkKafkaProducer<IN> extends RichSinkFunction<IN> {
* @param serializationSchema
* User defined (keyless) serialization schema.
*/
- public FlinkKafkaProducer(String brokerList, String topicId, SerializationSchema<IN, byte[]> serializationSchema) {
+ public FlinkKafkaProducer(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), null);
}
@@ -128,7 +128,7 @@ public class FlinkKafkaProducer<IN> extends RichSinkFunction<IN> {
* @param producerConfig
* Properties with the producer configuration.
*/
- public FlinkKafkaProducer(String topicId, SerializationSchema<IN, byte[]> serializationSchema, Properties producerConfig) {
+ public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) {
this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, null);
}
@@ -140,7 +140,7 @@ public class FlinkKafkaProducer<IN> extends RichSinkFunction<IN> {
* @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument.
* @param customPartitioner A serializable partitioner for assining messages to Kafka partitions.
*/
- public FlinkKafkaProducer(String topicId, SerializationSchema<IN, byte[]> serializationSchema, Properties producerConfig, KafkaPartitioner customPartitioner) {
+ public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner customPartitioner) {
this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java
index e832f20..afa2e42 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/api/KafkaSink.java
@@ -30,7 +30,7 @@ import org.apache.flink.streaming.util.serialization.SerializationSchema;
*/
@Deprecated
public class KafkaSink<IN> extends FlinkKafkaProducer<IN> {
- public KafkaSink(String brokerList, String topicId, SerializationSchema<IN, byte[]> serializationSchema) {
+ public KafkaSink(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) {
super(brokerList, topicId, serializationSchema);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTest.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTest.java
index e35fcfb..ec7db42 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTest.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTest.java
@@ -21,7 +21,7 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.commons.collections.map.LinkedMap;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
-import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.junit.Ignore;
import org.junit.Test;
@@ -132,7 +132,7 @@ public class KafkaConsumerTest {
props.setProperty("bootstrap.servers", "localhost:11111, localhost:22222");
props.setProperty("group.id", "non-existent-group");
- new FlinkKafkaConsumer<>("no op topic", new JavaDefaultStringSchema(), props,
+ new FlinkKafkaConsumer<>("no op topic", new SimpleStringSchema(), props,
FlinkKafkaConsumer.OffsetStore.FLINK_ZOOKEEPER,
FlinkKafkaConsumer.FetcherType.LEGACY_LOW_LEVEL);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index 2c48bea..48f4c50 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -64,7 +64,7 @@ import org.apache.flink.streaming.connectors.kafka.testutils.ThrottledMapper;
import org.apache.flink.streaming.connectors.kafka.testutils.Tuple2Partitioner;
import org.apache.flink.streaming.connectors.kafka.testutils.ValidatingExactlyOnceSink;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
-import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.util.serialization.TypeInformationKeyValueSerializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;
@@ -129,7 +129,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
public void runCheckpointingTest() throws Exception {
createTestTopic("testCheckpointing", 1, 1);
- FlinkKafkaConsumer<String> source = getConsumer("testCheckpointing", new JavaDefaultStringSchema(), standardProps);
+ FlinkKafkaConsumer<String> source = getConsumer("testCheckpointing", new SimpleStringSchema(), standardProps);
Field pendingCheckpointsField = FlinkKafkaConsumer.class.getDeclaredField("pendingCheckpoints");
pendingCheckpointsField.setAccessible(true);
LinkedMap pendingCheckpoints = (LinkedMap) pendingCheckpointsField.get(source);
@@ -577,7 +577,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env.enableCheckpointing(100);
env.getConfig().disableSysoutLogging();
- FlinkKafkaConsumer<String> source = getConsumer(topic, new JavaDefaultStringSchema(), standardProps);
+ FlinkKafkaConsumer<String> source = getConsumer(topic, new SimpleStringSchema(), standardProps);
env.addSource(source).addSink(new DiscardingSink<String>());
@@ -642,7 +642,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env.enableCheckpointing(100);
env.getConfig().disableSysoutLogging();
- FlinkKafkaConsumer<String> source = getConsumer(topic, new JavaDefaultStringSchema(), standardProps);
+ FlinkKafkaConsumer<String> source = getConsumer(topic, new SimpleStringSchema(), standardProps);
env.addSource(source).addSink(new DiscardingSink<String>());
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
index c5c3387..531b219 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTest.java
@@ -20,7 +20,7 @@ package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.connectors.kafka.testutils.MockRuntimeContext;
-import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.util.TestLogger;
import org.apache.kafka.clients.producer.Callback;
@@ -78,7 +78,7 @@ public class KafkaProducerTest extends TestLogger {
// (1) producer that propagates errors
FlinkKafkaProducer<String> producerPropagating = new FlinkKafkaProducer<String>(
- "mock_topic", new JavaDefaultStringSchema(), new Properties(), null);
+ "mock_topic", new SimpleStringSchema(), new Properties(), null);
producerPropagating.setRuntimeContext(new MockRuntimeContext(17, 3));
producerPropagating.open(new Configuration());
@@ -97,7 +97,7 @@ public class KafkaProducerTest extends TestLogger {
// (2) producer that only logs errors
FlinkKafkaProducer<String> producerLogging = new FlinkKafkaProducer<String>(
- "mock_topic", new JavaDefaultStringSchema(), new Properties(), null);
+ "mock_topic", new SimpleStringSchema(), new Properties(), null);
producerLogging.setLogFailuresOnly(true);
producerLogging.setRuntimeContext(new MockRuntimeContext(17, 3));
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
index 32377ae..22c6cfb 100644
--- a/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
+++ b/flink-streaming-connectors/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
@@ -28,7 +28,7 @@ import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner;
-import org.apache.flink.streaming.util.serialization.JavaDefaultStringSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.streaming.util.serialization.TypeInformationSerializationSchema;
import java.util.Random;
@@ -168,7 +168,7 @@ public class DataGenerators {
// we manually feed data into the Kafka sink
FlinkKafkaProducer<String> producer = null;
try {
- producer = new FlinkKafkaProducer<>(kafkaConnectionString, topic, new JavaDefaultStringSchema());
+ producer = new FlinkKafkaProducer<>(kafkaConnectionString, topic, new SimpleStringSchema());
producer.setRuntimeContext(new MockRuntimeContext(1,0));
producer.open(new Configuration());
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
index fa729d6..ca18fc4 100644
--- a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
+++ b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
@@ -39,9 +39,9 @@ public class RMQSink<IN> extends RichSinkFunction<IN> {
private transient ConnectionFactory factory;
private transient Connection connection;
private transient Channel channel;
- private SerializationSchema<IN, byte[]> schema;
+ private SerializationSchema<IN> schema;
- public RMQSink(String HOST_NAME, String QUEUE_NAME, SerializationSchema<IN, byte[]> schema) {
+ public RMQSink(String HOST_NAME, String QUEUE_NAME, SerializationSchema<IN> schema) {
this.HOST_NAME = HOST_NAME;
this.QUEUE_NAME = QUEUE_NAME;
this.schema = schema;
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQTopology.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQTopology.java b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQTopology.java
index 1f85862..debcce4 100644
--- a/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQTopology.java
+++ b/flink-streaming-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQTopology.java
@@ -19,7 +19,6 @@ package org.apache.flink.streaming.connectors.rabbitmq;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.util.serialization.SerializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
public class RMQTopology {
@@ -30,23 +29,14 @@ public class RMQTopology {
@SuppressWarnings("unused")
DataStreamSink<String> dataStream1 = env.addSource(
- new RMQSource<String>("localhost", "hello", new SimpleStringSchema())).print();
+ new RMQSource<>("localhost", "hello", new SimpleStringSchema())).print();
@SuppressWarnings("unused")
DataStreamSink<String> dataStream2 = env.fromElements("one", "two", "three", "four", "five",
"q").addSink(
- new RMQSink<String>("localhost", "hello", new StringToByteSerializer()));
+ new RMQSink<>("localhost", "hello", new SimpleStringSchema()));
env.execute();
}
- public static class StringToByteSerializer implements SerializationSchema<String, byte[]> {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public byte[] serialize(String element) {
- return element.getBytes();
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 309eb05..5084861 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -976,7 +976,7 @@ public class DataStream<T> {
* schema for serialization
* @return the closed DataStream
*/
- public DataStreamSink<T> writeToSocket(String hostName, int port, SerializationSchema<T, byte[]> schema) {
+ public DataStreamSink<T> writeToSocket(String hostName, int port, SerializationSchema<T> schema) {
DataStreamSink<T> returnStream = addSink(new SocketClientSink<T>(hostName, port, schema, 0));
returnStream.setParallelism(1); // It would not work if multiple instances would connect to the same port
return returnStream;
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
index 1356263..df9b124 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/SocketClientSink.java
@@ -51,7 +51,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
private final SerializableObject lock = new SerializableObject();
- private final SerializationSchema<IN, byte[]> schema;
+ private final SerializationSchema<IN> schema;
private final String hostName;
private final int port;
private final int maxNumRetries;
@@ -72,7 +72,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
* @param port Port of the server.
* @param schema Schema used to serialize the data into bytes.
*/
- public SocketClientSink(String hostName, int port, SerializationSchema<IN, byte[]> schema) {
+ public SocketClientSink(String hostName, int port, SerializationSchema<IN> schema) {
this(hostName, port, schema, 0);
}
@@ -86,7 +86,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
* @param schema Schema used to serialize the data into bytes.
* @param maxNumRetries The maximum number of retries after a message send failed.
*/
- public SocketClientSink(String hostName, int port, SerializationSchema<IN, byte[]> schema, int maxNumRetries) {
+ public SocketClientSink(String hostName, int port, SerializationSchema<IN> schema, int maxNumRetries) {
this(hostName, port, schema, maxNumRetries, false);
}
@@ -100,7 +100,7 @@ public class SocketClientSink<IN> extends RichSinkFunction<IN> {
* @param maxNumRetries The maximum number of retries after a message send failed.
* @param autoflush Flag to indicate whether the socket stream should be flushed after each message.
*/
- public SocketClientSink(String hostName, int port, SerializationSchema<IN, byte[]> schema,
+ public SocketClientSink(String hostName, int port, SerializationSchema<IN> schema,
int maxNumRetries, boolean autoflush)
{
checkArgument(port > 0 && port < 65536, "port is out of range");
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/JavaDefaultStringSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/JavaDefaultStringSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/JavaDefaultStringSchema.java
deleted file mode 100644
index ebb785c..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/JavaDefaultStringSchema.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.util.serialization;
-
-import org.apache.commons.lang3.SerializationUtils;
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-
-public class JavaDefaultStringSchema implements DeserializationSchema<String>, SerializationSchema<String, byte[]> {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public boolean isEndOfStream(String nextElement) {
- return nextElement.equals("q");
- }
-
- @Override
- public byte[] serialize(String element) {
- return SerializationUtils.serialize(element);
- }
-
- @Override
- public String deserialize(byte[] message) {
- return SerializationUtils.deserialize(message);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return BasicTypeInfo.STRING_TYPE_INFO;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchemaWrapper.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchemaWrapper.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchemaWrapper.java
index 26809aa..a1a8fc0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchemaWrapper.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/KeyedSerializationSchemaWrapper.java
@@ -25,9 +25,9 @@ public class KeyedSerializationSchemaWrapper<T> implements KeyedSerializationSch
private static final long serialVersionUID = 1351665280744549933L;
- private final SerializationSchema<T, byte[]> serializationSchema;
+ private final SerializationSchema<T> serializationSchema;
- public KeyedSerializationSchemaWrapper(SerializationSchema<T, byte[]> serializationSchema) {
+ public KeyedSerializationSchemaWrapper(SerializationSchema<T> serializationSchema) {
this.serializationSchema = serializationSchema;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/RawSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/RawSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/RawSchema.java
deleted file mode 100644
index 4d9aaee..0000000
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/RawSchema.java
+++ /dev/null
@@ -1,52 +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.util.serialization;
-
-import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
-import org.apache.flink.api.common.typeinfo.TypeInformation;
-
-/**
- * A "no-op" serialization and deserialization schema for byte strings. The serialized representation is
- * identical with the original representation.
- *
- * <p>This schema never considers a byte string to signal end-of-stream.</p>
- */
-public class RawSchema implements DeserializationSchema<byte[]>, SerializationSchema<byte[], byte[]> {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public byte[] deserialize(byte[] message) {
- return message;
- }
-
- @Override
- public boolean isEndOfStream(byte[] nextElement) {
- return false;
- }
-
- @Override
- public byte[] serialize(byte[] element) {
- return element;
- }
-
- @Override
- public TypeInformation<byte[]> getProducedType() {
- return PrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java
index 21342b2..a37c2f9 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SerializationSchema.java
@@ -25,9 +25,8 @@ import java.io.Serializable;
* to them in a specific format (for example as byte strings).
*
* @param <T> The type to be serialized.
- * @param <R> The serialized representation type.
*/
-public interface SerializationSchema<T, R> extends Serializable {
+public interface SerializationSchema<T> extends Serializable {
/**
* Serializes the incoming element to a specified type.
@@ -36,5 +35,5 @@ public interface SerializationSchema<T, R> extends Serializable {
* The incoming element to be serialized
* @return The serialized element.
*/
- R serialize(T element);
+ byte[] serialize(T element);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SimpleStringSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SimpleStringSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SimpleStringSchema.java
index 51d2d7f..2c6457d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SimpleStringSchema.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/SimpleStringSchema.java
@@ -20,14 +20,29 @@ package org.apache.flink.streaming.util.serialization;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import java.nio.charset.Charset;
+
+/**
+ * Very simple serialization schema for strings.
+ */
public class SimpleStringSchema implements DeserializationSchema<String>,
- SerializationSchema<String, String> {
+ SerializationSchema<String> {
private static final long serialVersionUID = 1L;
+ private Charset charset = Charset.defaultCharset();
+
+ public SimpleStringSchema() {
+
+ }
+
+ public SimpleStringSchema(Charset charset) {
+ this.charset = charset;
+ }
+
@Override
public String deserialize(byte[] message) {
- return new String(message);
+ return new String(message, charset);
}
@Override
@@ -36,8 +51,8 @@ public class SimpleStringSchema implements DeserializationSchema<String>,
}
@Override
- public String serialize(String element) {
- return element;
+ public byte[] serialize(String element) {
+ return element.getBytes(charset);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationSerializationSchema.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationSerializationSchema.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationSerializationSchema.java
index e937838..ce9c9ca 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationSerializationSchema.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/util/serialization/TypeInformationSerializationSchema.java
@@ -34,7 +34,7 @@ import java.io.IOException;
*
* @param <T> The type to be serialized.
*/
-public class TypeInformationSerializationSchema<T> implements DeserializationSchema<T>, SerializationSchema<T, byte[]> {
+public class TypeInformationSerializationSchema<T> implements DeserializationSchema<T>, SerializationSchema<T> {
private static final long serialVersionUID = -5359448468131559102L;
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
index 8f4acde..23491da 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java
@@ -52,7 +52,7 @@ public class SocketClientSinkTest extends TestLogger {
private static final String host = "127.0.0.1";
- private SerializationSchema<String, byte[]> simpleSchema = new SerializationSchema<String, byte[]>() {
+ private SerializationSchema<String> simpleSchema = new SerializationSchema<String>() {
@Override
public byte[] serialize(String element) {
return element.getBytes();
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/outputformat/SocketOutputFormatITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/outputformat/SocketOutputFormatITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/outputformat/SocketOutputFormatITCase.java
index 49876ec..07c7b74 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/outputformat/SocketOutputFormatITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/outputformat/SocketOutputFormatITCase.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.util.SocketOutputTestBase;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.test.testdata.WordCountData;
import org.junit.Ignore;
@@ -44,7 +45,7 @@ public class SocketOutputFormatITCase extends SocketOutputTestBase {
return value.toString() + "\n";
}
});
- counts.writeToSocket(HOST, port, new DummyStringSchema());
+ counts.writeToSocket(HOST, port, new SimpleStringSchema());
env.execute("WriteToSocketTest");
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java
index 4ded0fa..7d6a6d0 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/SocketOutputTestBase.java
@@ -21,6 +21,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.util.serialization.DeserializationSchema;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
import org.apache.flink.test.testdata.WordCountData;
import org.apache.flink.util.NetUtils;
@@ -78,7 +79,7 @@ public abstract class SocketOutputTestBase extends StreamingProgramTestBase {
public void waitForAccept() throws Exception {
Socket socket = serverSocket.accept();
BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
- DeserializationSchema<String> schema = new DummyStringSchema();
+ DeserializationSchema<String> schema = new SimpleStringSchema();
String rawData = in.readLine();
while (rawData != null){
String string = schema.deserialize(rawData.getBytes());
@@ -102,29 +103,4 @@ public abstract class SocketOutputTestBase extends StreamingProgramTestBase {
t.start();
}
}
-
- public static class DummyStringSchema implements DeserializationSchema<String>, SerializationSchema<String, byte[]>{
- private static final long serialVersionUID = 1L;
-
- @Override
- public boolean isEndOfStream(String nextElement) {
- return nextElement.equals("q");
- }
-
- @Override
- public byte[] serialize(String element) {
- return element.getBytes();
- }
-
- @Override
- public String deserialize(byte[] message) {
- return new String(message);
- }
-
- @Override
- public TypeInformation<String> getProducedType() {
- return TypeExtractor.getForClass(String.class);
- }
-
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index 6a3118a..e9cc041 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -725,7 +725,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
def writeToSocket(
hostname: String,
port: Integer,
- schema: SerializationSchema[T, Array[Byte]]): DataStreamSink[T] = {
+ schema: SerializationSchema[T]): DataStreamSink[T] = {
javaStream.writeToSocket(hostname, port, schema)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/09ea3dd0/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
index 2168bf7..36b1068 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
@@ -17,7 +17,8 @@
*/
package org.apache.flink.streaming.api.scala
-import org.apache.flink.streaming.util.SocketOutputTestBase.DummyStringSchema
+
+import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import scala.language.existentials
@@ -67,7 +68,7 @@ object OutputFormatTestPrograms {
.sum(1)
.map(tuple => tuple.toString() + "\n")
- counts.writeToSocket(outputHost, outputPort, new DummyStringSchema())
+ counts.writeToSocket(outputHost, outputPort, new SimpleStringSchema())
env.execute("Scala WordCountToCsv")
}