Return-Path: X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183]) by cust-asf2.ponee.io (Postfix) with ESMTP id 0C681200BCE for ; Fri, 2 Dec 2016 14:34:56 +0100 (CET) Received: by cust-asf.ponee.io (Postfix) id 0B439160B4A; Fri, 2 Dec 2016 13:34:56 +0000 (UTC) Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by cust-asf.ponee.io (Postfix) with SMTP id 3B506160B32 for ; Fri, 2 Dec 2016 14:34:53 +0100 (CET) Received: (qmail 70020 invoked by uid 500); 2 Dec 2016 13:34:52 -0000 Mailing-List: contact commits-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@flink.apache.org Delivered-To: mailing list commits@flink.apache.org Received: (qmail 69228 invoked by uid 99); 2 Dec 2016 13:34:51 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 02 Dec 2016 13:34:51 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 8780FF177F; Fri, 2 Dec 2016 13:34:51 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: fhueske@apache.org To: commits@flink.apache.org Date: Fri, 02 Dec 2016 13:35:03 -0000 Message-Id: In-Reply-To: <627ed550fa9e4aed8e3752516869c22e@git.apache.org> References: <627ed550fa9e4aed8e3752516869c22e@git.apache.org> X-Mailer: ASF-Git Admin Mailer Subject: [13/51] [abbrv] [partial] flink git commit: [FLINK-4676] [connectors] Merge batch and streaming connectors into common Maven module. archived-at: Fri, 02 Dec 2016 13:34:56 -0000 http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-flume/pom.xml ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-flume/pom.xml b/flink-streaming-connectors/flink-connector-flume/pom.xml deleted file mode 100644 index 1b1b810..0000000 --- a/flink-streaming-connectors/flink-connector-flume/pom.xml +++ /dev/null @@ -1,175 +0,0 @@ - - - - - 4.0.0 - - - org.apache.flink - flink-streaming-connectors - 1.2-SNAPSHOT - .. - - - flink-connector-flume_2.10 - flink-connector-flume - - jar - - - - 1.5.0 - - - - - - org.apache.flink - flink-streaming-java_2.10 - ${project.version} - provided - - - - org.apache.flume - flume-ng-core - ${flume-ng.version} - - - org.slf4j - slf4j-log4j12 - - - log4j - log4j - - - commons-io - commons-io - - - commons-codec - commons-codec - - - commons-cli - commons-cli - - - commons-lang - commons-lang - - - org.apache.avro - avro - - - org.codehaus.jackson - jackson-core-asl - - - org.codehaus.jackson - jackson-mapper-asl - - - com.thoughtworks.paranamer - paranamer - - - org.xerial.snappy - snappy-java - - - org.tukaani - xz - - - org.apache.velocity - velocity - - - commons-collections - commons-collections - - - org.mortbay.jetty - servlet-api - - - org.mortbay.jetty - jetty-util - - - org.mortbay.jetty - jetty - - - com.google.code.gson - gson - - - org.apache.thrift - libthrift - - - - - - - - - - org.apache.maven.plugins - maven-jar-plugin - - - - test-jar - - - - - - - - org.apache.maven.plugins - maven-shade-plugin - - - shade-flink - - - - - org.apache.flume:* - - - - - - - - - - - - http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/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 deleted file mode 100644 index 2dc043b..0000000 --- a/flink-streaming-connectors/flink-connector-flume/src/main/java/org/apache/flink/streaming/connectors/flume/FlumeSink.java +++ /dev/null @@ -1,141 +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.flume; - -import org.apache.flink.configuration.Configuration; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; -import org.apache.flink.streaming.util.serialization.SerializationSchema; -import org.apache.flume.Event; -import org.apache.flume.EventDeliveryException; -import org.apache.flume.FlumeException; -import org.apache.flume.api.RpcClient; -import org.apache.flume.api.RpcClientFactory; -import org.apache.flume.event.EventBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class FlumeSink extends RichSinkFunction { - private static final long serialVersionUID = 1L; - - private static final Logger LOG = LoggerFactory.getLogger(FlumeSink.class); - - private transient FlinkRpcClientFacade client; - boolean initDone = false; - String host; - int port; - SerializationSchema schema; - - public FlumeSink(String host, int port, SerializationSchema schema) { - this.host = host; - this.port = port; - this.schema = schema; - } - - /** - * Receives tuples from the Apache Flink {@link DataStream} and forwards - * them to Apache Flume. - * - * @param value - * The tuple arriving from the datastream - */ - @Override - public void invoke(IN value) { - - byte[] data = schema.serialize(value); - client.sendDataToFlume(data); - - } - - private class FlinkRpcClientFacade { - private RpcClient client; - private String hostname; - private int port; - - /** - * Initializes the connection to Apache Flume. - * - * @param hostname - * The host - * @param port - * The port. - */ - public void init(String hostname, int port) { - // Setup the RPC connection - this.hostname = hostname; - this.port = port; - int initCounter = 0; - while (true) { - if (initCounter >= 90) { - throw new RuntimeException("Cannot establish connection with" + port + " at " - + host); - } - try { - this.client = RpcClientFactory.getDefaultInstance(hostname, port); - } catch (FlumeException e) { - // Wait one second if the connection failed before the next - // try - try { - Thread.sleep(1000); - } catch (InterruptedException e1) { - if (LOG.isErrorEnabled()) { - LOG.error("Interrupted while trying to connect {} at {}", port, host); - } - } - } - if (client != null) { - break; - } - initCounter++; - } - initDone = true; - } - - /** - * Sends byte arrays as {@link Event} series to Apache Flume. - * - * @param data - * The byte array to send to Apache FLume - */ - public void sendDataToFlume(byte[] data) { - Event event = EventBuilder.withBody(data); - - try { - client.append(event); - - } catch (EventDeliveryException e) { - // clean up and recreate the client - client.close(); - client = null; - client = RpcClientFactory.getDefaultInstance(hostname, port); - } - } - - } - - @Override - public void close() { - client.client.close(); - } - - @Override - public void open(Configuration config) { - client = new FlinkRpcClientFacade(); - client.init(host, port); - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml b/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml deleted file mode 100644 index 04019f8..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/pom.xml +++ /dev/null @@ -1,205 +0,0 @@ - - - - - 4.0.0 - - - org.apache.flink - flink-streaming-connectors - 1.2-SNAPSHOT - .. - - - flink-connector-kafka-0.10_2.10 - flink-connector-kafka-0.10 - - jar - - - - 0.10.0.1 - - - - - - - - org.apache.flink - flink-connector-kafka-0.9_2.10 - ${project.version} - - - org.apache.kafka - kafka_${scala.binary.version} - - - - - - - - org.apache.kafka - kafka-clients - ${kafka.version} - - - - org.apache.flink - flink-table_2.10 - ${project.version} - provided - - true - - - - - - org.apache.flink - flink-streaming-java_2.10 - ${project.version} - test - test-jar - - - - org.apache.flink - flink-connector-kafka-0.9_2.10 - ${project.version} - - - - org.apache.kafka - kafka_${scala.binary.version} - - - test-jar - test - - - - org.apache.flink - flink-connector-kafka-base_2.10 - ${project.version} - - - - org.apache.kafka - kafka_${scala.binary.version} - - - test-jar - test - - - - - org.apache.kafka - kafka_${scala.binary.version} - ${kafka.version} - test - - - - org.apache.flink - flink-tests_2.10 - ${project.version} - test-jar - test - - - - org.apache.flink - flink-test-utils_2.10 - ${project.version} - test - - - - org.apache.flink - flink-runtime_2.10 - ${project.version} - test-jar - test - - - - org.apache.flink - flink-metrics-jmx - ${project.version} - test - - - - - - - - - org.apache.maven.plugins - maven-jar-plugin - - - - test-jar - - - - **/KafkaTestEnvironmentImpl* - - - - - - - org.apache.maven.plugins - maven-source-plugin - - - attach-test-sources - - test-jar-no-fork - - - - **/KafkaTestEnvironmentImpl* - - - - - - - org.apache.maven.plugins - maven-surefire-plugin - - - 1 - -Xms256m -Xmx1000m -Dlog4j.configuration=${log4j.configuration} -Dmvn.forkNumber=${surefire.forkNumber} -XX:-UseGCOverheadLimit - - - - - - http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java deleted file mode 100644 index a9ce336..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer010.java +++ /dev/null @@ -1,153 +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 org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks; -import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks; -import org.apache.flink.streaming.api.operators.StreamingRuntimeContext; -import org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher; -import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher; -import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition; -import org.apache.flink.streaming.util.serialization.DeserializationSchema; -import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema; -import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchemaWrapper; -import org.apache.flink.util.SerializedValue; - -import java.util.Collections; -import java.util.List; -import java.util.Properties; - - -/** - * The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from - * Apache Kafka 0.10.x. The consumer can run in multiple parallel instances, each of which will pull - * data from one or more Kafka partitions. - * - *

The Flink Kafka Consumer participates in checkpointing and guarantees that no data is lost - * during a failure, and that the computation processes elements "exactly once". - * (Note: These guarantees naturally assume that Kafka itself does not loose any data.)

- * - *

Please note that Flink snapshots the offsets internally as part of its distributed checkpoints. The offsets - * committed to Kafka / ZooKeeper are only to bring the outside view of progress in sync with Flink's view - * of the progress. That way, monitoring and other jobs can get a view of how far the Flink Kafka consumer - * has consumed a topic.

- * - *

Please refer to Kafka's documentation for the available configuration properties: - * http://kafka.apache.org/documentation.html#newconsumerconfigs

- * - *

NOTE: The implementation currently accesses partition metadata when the consumer - * is constructed. That means that the client that submits the program needs to be able to - * reach the Kafka brokers or ZooKeeper.

- */ -public class FlinkKafkaConsumer010 extends FlinkKafkaConsumer09 { - - private static final long serialVersionUID = 2324564345203409112L; - - - // ------------------------------------------------------------------------ - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.10.x - * - * @param topic - * The name of the topic that should be consumed. - * @param valueDeserializer - * The de-/serializer used to convert between Kafka's byte messages and Flink's objects. - * @param props - * The properties used to configure the Kafka consumer client, and the ZooKeeper client. - */ - public FlinkKafkaConsumer010(String topic, DeserializationSchema valueDeserializer, Properties props) { - this(Collections.singletonList(topic), valueDeserializer, props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.10.x - * - * This constructor allows passing a {@see KeyedDeserializationSchema} for reading key/value - * pairs, offsets, and topic names from Kafka. - * - * @param topic - * The name of the topic that should be consumed. - * @param deserializer - * The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects. - * @param props - * The properties used to configure the Kafka consumer client, and the ZooKeeper client. - */ - public FlinkKafkaConsumer010(String topic, KeyedDeserializationSchema deserializer, Properties props) { - this(Collections.singletonList(topic), deserializer, props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.10.x - * - * This constructor allows passing multiple topics to the consumer. - * - * @param topics - * The Kafka topics to read from. - * @param deserializer - * The de-/serializer used to convert between Kafka's byte messages and Flink's objects. - * @param props - * The properties that are used to configure both the fetcher and the offset handler. - */ - public FlinkKafkaConsumer010(List topics, DeserializationSchema deserializer, Properties props) { - this(topics, new KeyedDeserializationSchemaWrapper<>(deserializer), props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.10.x - * - * This constructor allows passing multiple topics and a key/value deserialization schema. - * - * @param topics - * The Kafka topics to read from. - * @param deserializer - * The keyed de-/serializer used to convert between Kafka's byte messages and Flink's objects. - * @param props - * The properties that are used to configure both the fetcher and the offset handler. - */ - public FlinkKafkaConsumer010(List topics, KeyedDeserializationSchema deserializer, Properties props) { - super(topics, deserializer, props); - } - - @Override - protected AbstractFetcher createFetcher( - SourceContext sourceContext, - List thisSubtaskPartitions, - SerializedValue> watermarksPeriodic, - SerializedValue> watermarksPunctuated, - StreamingRuntimeContext runtimeContext) throws Exception { - - boolean useMetrics = !Boolean.valueOf(properties.getProperty(KEY_DISABLE_METRICS, "false")); - - return new Kafka010Fetcher<>( - sourceContext, - thisSubtaskPartitions, - watermarksPeriodic, - watermarksPunctuated, - runtimeContext.getProcessingTimeService(), - runtimeContext.getExecutionConfig().getAutoWatermarkInterval(), - runtimeContext.getUserCodeClassLoader(), - runtimeContext.isCheckpointingEnabled(), - runtimeContext.getTaskNameWithSubtasks(), - runtimeContext.getMetricGroup(), - deserializer, - properties, - pollTimeout, - useMetrics); - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java deleted file mode 100644 index cc0194b..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer010.java +++ /dev/null @@ -1,398 +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 org.apache.flink.api.common.functions.IterationRuntimeContext; -import org.apache.flink.api.common.functions.RichFunction; -import org.apache.flink.api.common.functions.RuntimeContext; -import org.apache.flink.api.java.typeutils.GenericTypeInfo; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.datastream.DataStreamSink; -import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; -import org.apache.flink.streaming.api.functions.sink.SinkFunction; -import org.apache.flink.streaming.api.operators.StreamSink; -import org.apache.flink.streaming.connectors.kafka.partitioner.FixedPartitioner; -import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner; -import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; -import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema; -import org.apache.flink.streaming.util.serialization.KeyedSerializationSchemaWrapper; -import org.apache.flink.streaming.util.serialization.SerializationSchema; -import org.apache.kafka.clients.producer.ProducerRecord; - -import java.util.Properties; - -import static org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.getPropertiesFromBrokerList; - - -/** - * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.10.x - * - * Implementation note: This producer is a hybrid between a regular regular sink function (a) - * and a custom operator (b). - * - * For (a), the class implements the SinkFunction and RichFunction interfaces. - * For (b), it extends the StreamTask class. - * - * Details about approach (a): - * - * Pre Kafka 0.10 producers only follow approach (a), allowing users to use the producer using the - * DataStream.addSink() method. - * Since the APIs exposed in that variant do not allow accessing the the timestamp attached to the record - * the Kafka 0.10 producer has a second invocation option, approach (b). - * - * Details about approach (b): - * Kafka 0.10 supports writing the timestamp attached to a record to Kafka. When adding the - * FlinkKafkaProducer010 using the FlinkKafkaProducer010.writeToKafkaWithTimestamps() method, the Kafka producer - * can access the internal record timestamp of the record and write it to Kafka. - * - * All methods and constructors in this class are marked with the approach they are needed for. - */ -public class FlinkKafkaProducer010 extends StreamSink implements SinkFunction, RichFunction { - - /** - * Flag controlling whether we are writing the Flink record's timestamp into Kafka. - */ - private boolean writeTimestampToKafka = false; - - // ---------------------- "Constructors" for timestamp writing ------------------ - - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to - * the topic. - * - * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above) - * - * @param inStream The stream to write to Kafka - * @param topicId ID of the Kafka topic. - * @param serializationSchema User defined serialization schema supporting key/value messages - * @param producerConfig Properties with the producer configuration. - */ - public static FlinkKafkaProducer010Configuration writeToKafkaWithTimestamps(DataStream inStream, - String topicId, - KeyedSerializationSchema serializationSchema, - Properties producerConfig) { - return writeToKafkaWithTimestamps(inStream, topicId, serializationSchema, producerConfig, new FixedPartitioner()); - } - - - /** - * Creates a FlinkKafkaProducer for a given topic. the sink produces a DataStream to - * the topic. - * - * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above) - * - * @param inStream The stream to write to Kafka - * @param topicId ID of the Kafka topic. - * @param serializationSchema User defined (keyless) serialization schema. - * @param producerConfig Properties with the producer configuration. - */ - public static FlinkKafkaProducer010Configuration writeToKafkaWithTimestamps(DataStream inStream, - String topicId, - SerializationSchema serializationSchema, - Properties producerConfig) { - return writeToKafkaWithTimestamps(inStream, topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner()); - } - - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces a DataStream to - * the topic. - * - * This constructor allows writing timestamps to Kafka, it follow approach (b) (see above) - * - * @param inStream The stream to write to Kafka - * @param topicId The name of the target topic - * @param serializationSchema A serializable serialization schema for turning user objects into a kafka-consumable byte[] supporting key/value messages - * @param producerConfig Configuration properties for the KafkaProducer. 'bootstrap.servers.' is the only required argument. - * @param customPartitioner A serializable partitioner for assigning messages to Kafka partitions. - */ - public static FlinkKafkaProducer010Configuration writeToKafkaWithTimestamps(DataStream inStream, - String topicId, - KeyedSerializationSchema serializationSchema, - Properties producerConfig, - KafkaPartitioner customPartitioner) { - - GenericTypeInfo