http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java deleted file mode 100644 index 42b9682..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka010ProducerITCase.java +++ /dev/null @@ -1,33 +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.junit.Test; - - -@SuppressWarnings("serial") -public class Kafka010ProducerITCase extends KafkaProducerTestBase { - - @Test - public void testCustomPartitioning() { - runCustomPartitioningTest(); - } - -}
http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java deleted file mode 100644 index f15fd45..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestEnvironmentImpl.java +++ /dev/null @@ -1,420 +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 kafka.admin.AdminUtils; -import kafka.common.KafkaException; -import kafka.server.KafkaConfig; -import kafka.server.KafkaServer; -import kafka.utils.SystemTime$; -import kafka.utils.ZkUtils; -import org.I0Itec.zkclient.ZkClient; -import org.apache.commons.io.FileUtils; -import org.apache.curator.test.TestingServer; -import org.apache.flink.streaming.api.datastream.DataStream; -import org.apache.flink.streaming.api.datastream.DataStreamSink; -import org.apache.flink.streaming.api.operators.StreamSink; -import org.apache.flink.streaming.connectors.kafka.testutils.ZooKeeperStringSerializer; -import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner; -import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema; -import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema; -import org.apache.flink.util.NetUtils; -import org.apache.kafka.clients.consumer.KafkaConsumer; -import org.apache.kafka.clients.consumer.OffsetAndMetadata; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.protocol.SecurityProtocol; -import org.apache.kafka.common.requests.MetadataResponse; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.net.BindException; -import java.util.ArrayList; -import java.util.List; -import java.util.Properties; -import java.util.UUID; - -import static org.apache.flink.util.NetUtils.hostAndPortToUrlString; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; - -/** - * An implementation of the KafkaServerProvider for Kafka 0.10 - */ -public class KafkaTestEnvironmentImpl extends KafkaTestEnvironment { - - protected static final Logger LOG = LoggerFactory.getLogger(KafkaTestEnvironmentImpl.class); - private File tmpZkDir; - private File tmpKafkaParent; - private List<File> tmpKafkaDirs; - private List<KafkaServer> brokers; - private TestingServer zookeeper; - private String zookeeperConnectionString; - private String brokerConnectionString = ""; - private Properties standardProps; - private Properties additionalServerProperties; - private boolean secureMode = false; - // 6 seconds is default. Seems to be too small for travis. 30 seconds - private int zkTimeout = 30000; - - public String getBrokerConnectionString() { - return brokerConnectionString; - } - - @Override - public Properties getStandardProperties() { - return standardProps; - } - - @Override - public Properties getSecureProperties() { - Properties prop = new Properties(); - if(secureMode) { - prop.put("security.inter.broker.protocol", "SASL_PLAINTEXT"); - prop.put("security.protocol", "SASL_PLAINTEXT"); - prop.put("sasl.kerberos.service.name", "kafka"); - - //add special timeout for Travis - prop.setProperty("zookeeper.session.timeout.ms", String.valueOf(zkTimeout)); - prop.setProperty("zookeeper.connection.timeout.ms", String.valueOf(zkTimeout)); - prop.setProperty("metadata.fetch.timeout.ms","120000"); - } - return prop; - } - - @Override - public String getVersion() { - return "0.10"; - } - - @Override - public List<KafkaServer> getBrokers() { - return brokers; - } - - @Override - public <T> FlinkKafkaConsumerBase<T> getConsumer(List<String> topics, KeyedDeserializationSchema<T> readSchema, Properties props) { - return new FlinkKafkaConsumer010<>(topics, readSchema, props); - } - - @Override - public <T> StreamSink<T> getProducerSink(String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) { - FlinkKafkaProducer010<T> prod = new FlinkKafkaProducer010<>(topic, serSchema, props, partitioner); - prod.setFlushOnCheckpoint(true); - return new StreamSink<>(prod); - } - - - @Override - public <T> DataStreamSink<T> produceIntoKafka(DataStream<T> stream, String topic, KeyedSerializationSchema<T> serSchema, Properties props, KafkaPartitioner<T> partitioner) { - FlinkKafkaProducer010<T> prod = new FlinkKafkaProducer010<>(topic, serSchema, props, partitioner); - prod.setFlushOnCheckpoint(true); - return stream.addSink(prod); - } - - @Override - public KafkaOffsetHandler createOffsetHandler(Properties props) { - return new KafkaOffsetHandlerImpl(props); - } - - @Override - public void restartBroker(int leaderId) throws Exception { - brokers.set(leaderId, getKafkaServer(leaderId, tmpKafkaDirs.get(leaderId))); - } - - @Override - public int getLeaderToShutDown(String topic) throws Exception { - ZkUtils zkUtils = getZkUtils(); - try { - MetadataResponse.PartitionMetadata firstPart = null; - do { - if (firstPart != null) { - LOG.info("Unable to find leader. error code {}", firstPart.error().code()); - // not the first try. Sleep a bit - Thread.sleep(150); - } - - List<MetadataResponse.PartitionMetadata> partitionMetadata = AdminUtils.fetchTopicMetadataFromZk(topic, zkUtils).partitionMetadata(); - firstPart = partitionMetadata.get(0); - } - while (firstPart.error().code() != 0); - - return firstPart.leader().id(); - } finally { - zkUtils.close(); - } - } - - @Override - public int getBrokerId(KafkaServer server) { - return server.config().brokerId(); - } - - @Override - public boolean isSecureRunSupported() { - return true; - } - - @Override - public void prepare(int numKafkaServers, Properties additionalServerProperties, boolean secureMode) { - //increase the timeout since in Travis ZK connection takes long time for secure connection. - if(secureMode) { - //run only one kafka server to avoid multiple ZK connections from many instances - Travis timeout - numKafkaServers = 1; - zkTimeout = zkTimeout * 15; - } - - this.additionalServerProperties = additionalServerProperties; - this.secureMode = secureMode; - File tempDir = new File(System.getProperty("java.io.tmpdir")); - - tmpZkDir = new File(tempDir, "kafkaITcase-zk-dir-" + (UUID.randomUUID().toString())); - assertTrue("cannot create zookeeper temp dir", tmpZkDir.mkdirs()); - - tmpKafkaParent = new File(tempDir, "kafkaITcase-kafka-dir*" + (UUID.randomUUID().toString())); - assertTrue("cannot create kafka temp dir", tmpKafkaParent.mkdirs()); - - tmpKafkaDirs = new ArrayList<>(numKafkaServers); - for (int i = 0; i < numKafkaServers; i++) { - File tmpDir = new File(tmpKafkaParent, "server-" + i); - assertTrue("cannot create kafka temp dir", tmpDir.mkdir()); - tmpKafkaDirs.add(tmpDir); - } - - zookeeper = null; - brokers = null; - - try { - zookeeper = new TestingServer(- 1, tmpZkDir); - zookeeperConnectionString = zookeeper.getConnectString(); - LOG.info("Starting Zookeeper with zookeeperConnectionString: {}", zookeeperConnectionString); - - LOG.info("Starting KafkaServer"); - brokers = new ArrayList<>(numKafkaServers); - - for (int i = 0; i < numKafkaServers; i++) { - brokers.add(getKafkaServer(i, tmpKafkaDirs.get(i))); - - if(secureMode) { - brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.SASL_PLAINTEXT)) + ","; - } else { - brokerConnectionString += hostAndPortToUrlString(KafkaTestEnvironment.KAFKA_HOST, brokers.get(i).socketServer().boundPort(SecurityProtocol.PLAINTEXT)) + ","; - } - } - - LOG.info("ZK and KafkaServer started."); - } - catch (Throwable t) { - t.printStackTrace(); - fail("Test setup failed: " + t.getMessage()); - } - - standardProps = new Properties(); - standardProps.setProperty("zookeeper.connect", zookeeperConnectionString); - standardProps.setProperty("bootstrap.servers", brokerConnectionString); - standardProps.setProperty("group.id", "flink-tests"); - standardProps.setProperty("enable.auto.commit", "false"); - standardProps.setProperty("zookeeper.session.timeout.ms", String.valueOf(zkTimeout)); - standardProps.setProperty("zookeeper.connection.timeout.ms", String.valueOf(zkTimeout)); - standardProps.setProperty("auto.offset.reset", "earliest"); // read from the beginning. (earliest is kafka 0.10 value) - standardProps.setProperty("max.partition.fetch.bytes", "256"); // make a lot of fetches (MESSAGES MUST BE SMALLER!) - } - - @Override - public void shutdown() { - for (KafkaServer broker : brokers) { - if (broker != null) { - broker.shutdown(); - } - } - brokers.clear(); - - if (zookeeper != null) { - try { - zookeeper.stop(); - } - catch (Exception e) { - LOG.warn("ZK.stop() failed", e); - } - zookeeper = null; - } - - // clean up the temp spaces - - if (tmpKafkaParent != null && tmpKafkaParent.exists()) { - try { - FileUtils.deleteDirectory(tmpKafkaParent); - } - catch (Exception e) { - // ignore - } - } - if (tmpZkDir != null && tmpZkDir.exists()) { - try { - FileUtils.deleteDirectory(tmpZkDir); - } - catch (Exception e) { - // ignore - } - } - } - - public ZkUtils getZkUtils() { - ZkClient creator = new ZkClient(zookeeperConnectionString, Integer.valueOf(standardProps.getProperty("zookeeper.session.timeout.ms")), - Integer.valueOf(standardProps.getProperty("zookeeper.connection.timeout.ms")), new ZooKeeperStringSerializer()); - return ZkUtils.apply(creator, false); - } - - @Override - public void createTestTopic(String topic, int numberOfPartitions, int replicationFactor, Properties topicConfig) { - // create topic with one client - LOG.info("Creating topic {}", topic); - - ZkUtils zkUtils = getZkUtils(); - try { - AdminUtils.createTopic(zkUtils, topic, numberOfPartitions, replicationFactor, topicConfig, kafka.admin.RackAwareMode.Enforced$.MODULE$); - } finally { - zkUtils.close(); - } - - // validate that the topic has been created - final long deadline = System.currentTimeMillis() + 30000; - do { - try { - if(secureMode) { - //increase wait time since in Travis ZK timeout occurs frequently - int wait = zkTimeout / 100; - LOG.info("waiting for {} msecs before the topic {} can be checked", wait, topic); - Thread.sleep(wait); - } else { - Thread.sleep(100); - } - } catch (InterruptedException e) { - // restore interrupted state - } - // we could use AdminUtils.topicExists(zkUtils, topic) here, but it's results are - // not always correct. - - // create a new ZK utils connection - ZkUtils checkZKConn = getZkUtils(); - if(AdminUtils.topicExists(checkZKConn, topic)) { - checkZKConn.close(); - return; - } - checkZKConn.close(); - } - while (System.currentTimeMillis() < deadline); - fail("Test topic could not be created"); - } - - @Override - public void deleteTestTopic(String topic) { - ZkUtils zkUtils = getZkUtils(); - try { - LOG.info("Deleting topic {}", topic); - - ZkClient zk = new ZkClient(zookeeperConnectionString, Integer.valueOf(standardProps.getProperty("zookeeper.session.timeout.ms")), - Integer.valueOf(standardProps.getProperty("zookeeper.connection.timeout.ms")), new ZooKeeperStringSerializer()); - - AdminUtils.deleteTopic(zkUtils, topic); - - zk.close(); - } finally { - zkUtils.close(); - } - } - - /** - * Copied from com.github.sakserv.minicluster.KafkaLocalBrokerIntegrationTest (ASL licensed) - */ - protected KafkaServer getKafkaServer(int brokerId, File tmpFolder) throws Exception { - Properties kafkaProperties = new Properties(); - - // properties have to be Strings - kafkaProperties.put("advertised.host.name", KAFKA_HOST); - kafkaProperties.put("broker.id", Integer.toString(brokerId)); - kafkaProperties.put("log.dir", tmpFolder.toString()); - kafkaProperties.put("zookeeper.connect", zookeeperConnectionString); - kafkaProperties.put("message.max.bytes", String.valueOf(50 * 1024 * 1024)); - kafkaProperties.put("replica.fetch.max.bytes", String.valueOf(50 * 1024 * 1024)); - - // for CI stability, increase zookeeper session timeout - kafkaProperties.put("zookeeper.session.timeout.ms", zkTimeout); - kafkaProperties.put("zookeeper.connection.timeout.ms", zkTimeout); - if(additionalServerProperties != null) { - kafkaProperties.putAll(additionalServerProperties); - } - - final int numTries = 5; - - for (int i = 1; i <= numTries; i++) { - int kafkaPort = NetUtils.getAvailablePort(); - kafkaProperties.put("port", Integer.toString(kafkaPort)); - - //to support secure kafka cluster - if(secureMode) { - LOG.info("Adding Kafka secure configurations"); - kafkaProperties.put("listeners", "SASL_PLAINTEXT://" + KAFKA_HOST + ":" + kafkaPort); - kafkaProperties.put("advertised.listeners", "SASL_PLAINTEXT://" + KAFKA_HOST + ":" + kafkaPort); - kafkaProperties.putAll(getSecureProperties()); - } - - KafkaConfig kafkaConfig = new KafkaConfig(kafkaProperties); - - try { - scala.Option<String> stringNone = scala.Option.apply(null); - KafkaServer server = new KafkaServer(kafkaConfig, SystemTime$.MODULE$, stringNone); - server.startup(); - return server; - } - catch (KafkaException e) { - if (e.getCause() instanceof BindException) { - // port conflict, retry... - LOG.info("Port conflict when starting Kafka Broker. Retrying..."); - } - else { - throw e; - } - } - } - - throw new Exception("Could not start Kafka after " + numTries + " retries due to port conflicts."); - } - - private class KafkaOffsetHandlerImpl implements KafkaOffsetHandler { - - private final KafkaConsumer<byte[], byte[]> offsetClient; - - public KafkaOffsetHandlerImpl(Properties props) { - offsetClient = new KafkaConsumer<>(props); - } - - @Override - public Long getCommittedOffset(String topicName, int partition) { - OffsetAndMetadata committed = offsetClient.committed(new TopicPartition(topicName, partition)); - return (committed != null) ? committed.offset() : null; - } - - @Override - public void close() { - offsetClient.close(); - } - } - -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties deleted file mode 100644 index fbeb110..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/log4j-test.properties +++ /dev/null @@ -1,30 +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. -################################################################################ - -log4j.rootLogger=INFO, 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 -log4j.logger.org.apache.zookeeper=OFF, testlogger -log4j.logger.state.change.logger=OFF, testlogger -log4j.logger.kafka=OFF, testlogger http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml b/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml deleted file mode 100644 index 45b3b92..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.10/src/test/resources/logback-test.xml +++ /dev/null @@ -1,30 +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. - --> - -<configuration> - <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> - <encoder> - <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} - %msg%n</pattern> - </encoder> - </appender> - - <root level="WARN"> - <appender-ref ref="STDOUT"/> - </root> - <logger name="org.apache.flink.streaming" level="WARN"/> -</configuration> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/pom.xml ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/pom.xml b/flink-streaming-connectors/flink-connector-kafka-0.8/pom.xml deleted file mode 100644 index f17f9ae..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/pom.xml +++ /dev/null @@ -1,219 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<!-- -Licensed to the Apache Software Foundation (ASF) under one -or more contributor license agreements. See the NOTICE file -distributed with this work for additional information -regarding copyright ownership. The ASF licenses this file -to you under the Apache License, Version 2.0 (the -"License"); you may not use this file except in compliance -with the License. You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, -software distributed under the License is distributed on an -"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -KIND, either express or implied. See the License for the -specific language governing permissions and limitations -under the License. ---> -<project xmlns="http://maven.apache.org/POM/4.0.0"; - xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; - xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd";> - - <modelVersion>4.0.0</modelVersion> - - <parent> - <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-connectors</artifactId> - <version>1.2-SNAPSHOT</version> - <relativePath>..</relativePath> - </parent> - - <artifactId>flink-connector-kafka-0.8_2.10</artifactId> - <name>flink-connector-kafka-0.8</name> - - <packaging>jar</packaging> - - <!-- Allow users to pass custom connector versions --> - <properties> - <kafka.version>0.8.2.2</kafka.version> - </properties> - - <dependencies> - - <!-- core dependencies --> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_2.10</artifactId> - <version>${project.version}</version> - <scope>provided</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-shaded-curator-recipes</artifactId> - <version>${project.version}</version> - <scope>provided</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-connector-kafka-base_2.10</artifactId> - <version>${project.version}</version> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-table_2.10</artifactId> - <version>${project.version}</version> - <scope>provided</scope> - <!-- Projects depending on this project, - won't depend on flink-table. --> - <optional>true</optional> - </dependency> - - <dependency> - <groupId>org.apache.kafka</groupId> - <artifactId>kafka_${scala.binary.version}</artifactId> - <version>${kafka.version}</version> - <exclusions> - <exclusion> - <groupId>com.sun.jmx</groupId> - <artifactId>jmxri</artifactId> - </exclusion> - <exclusion> - <groupId>com.sun.jdmk</groupId> - <artifactId>jmxtools</artifactId> - </exclusion> - <exclusion> - <groupId>log4j</groupId> - <artifactId>log4j</artifactId> - </exclusion> - <exclusion> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-simple</artifactId> - </exclusion> - <exclusion> - <groupId>net.sf.jopt-simple</groupId> - <artifactId>jopt-simple</artifactId> - </exclusion> - <exclusion> - <groupId>org.scala-lang</groupId> - <artifactId>scala-reflect</artifactId> - </exclusion> - <exclusion> - <groupId>org.scala-lang</groupId> - <artifactId>scala-compiler</artifactId> - </exclusion> - <exclusion> - <groupId>com.yammer.metrics</groupId> - <artifactId>metrics-annotation</artifactId> - </exclusion> - <exclusion> - <groupId>org.xerial.snappy</groupId> - <artifactId>snappy-java</artifactId> - </exclusion> - </exclusions> - </dependency> - - <!-- test dependencies --> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-streaming-java_2.10</artifactId> - <version>${project.version}</version> - <scope>test</scope> - <type>test-jar</type> - </dependency> - - <dependency> - <groupId>org.apache.curator</groupId> - <artifactId>curator-test</artifactId> - <version>${curator.version}</version> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-metrics-jmx</artifactId> - <version>${project.version}</version> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-connector-kafka-base_2.10</artifactId> - <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-runtime_2.10</artifactId> - <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-test-utils_2.10</artifactId> - <version>${project.version}</version> - <scope>test</scope> - </dependency> - - <dependency> - <groupId>org.apache.flink</groupId> - <artifactId>flink-tests_2.10</artifactId> - <version>${project.version}</version> - <type>test-jar</type> - <scope>test</scope> - </dependency> - - </dependencies> - - - <build> - <plugins> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-surefire-plugin</artifactId> - <configuration> - <!-- Enforce single fork execution due to heavy mini cluster use in the tests --> - <forkCount>1</forkCount> - </configuration> - </plugin> - <!-- Relocate curator --> - <plugin> - <groupId>org.apache.maven.plugins</groupId> - <artifactId>maven-shade-plugin</artifactId> - <executions> - <execution> - <id>shade-flink</id> - <phase>package</phase> - <goals> - <goal>shade</goal> - </goals> - <configuration> - <artifactSet> - <includes combine.children="append"> - <include>org.apache.flink:flink-shaded-curator-recipes</include> - </includes> - </artifactSet> - <relocations combine.children="append"> - <relocation> - <pattern>org.apache.curator</pattern> - <shadedPattern>org.apache.flink.shaded.org.apache.curator</shadedPattern> - </relocation> - </relocations> - </configuration> - </execution> - </executions> - </plugin> - </plugins> - </build> - -</project> http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.java deleted file mode 100644 index 0aacccd..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer08.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 kafka.api.OffsetRequest; -import kafka.cluster.Broker; -import kafka.common.ErrorMapping; -import kafka.javaapi.PartitionMetadata; -import kafka.javaapi.TopicMetadata; -import kafka.javaapi.TopicMetadataRequest; -import kafka.javaapi.consumer.SimpleConsumer; - -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.internals.AbstractFetcher; -import org.apache.flink.streaming.connectors.kafka.internals.Kafka08Fetcher; -import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition; -import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartitionLeader; -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.NetUtils; -import org.apache.flink.util.PropertiesUtil; -import org.apache.flink.util.SerializedValue; - -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.common.Node; - -import java.net.InetAddress; -import java.net.URL; -import java.net.UnknownHostException; -import java.nio.channels.ClosedChannelException; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Properties; -import java.util.Random; - -import static org.apache.flink.util.PropertiesUtil.getInt; -import static org.apache.flink.util.Preconditions.checkNotNull; - -/** - * The Flink Kafka Consumer is a streaming data source that pulls a parallel data stream from - * Apache Kafka 0.8.x. The consumer can run in multiple parallel instances, each of which will pull - * data from one or more Kafka partitions. - * - * <p>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.)</p> - * - * <p>Flink's Kafka Consumer is designed to be compatible with Kafka's High-Level Consumer API (0.8.x). - * Most of Kafka's configuration variables can be used with this consumer as well: - * <ul> - * <li>socket.timeout.ms</li> - * <li>socket.receive.buffer.bytes</li> - * <li>fetch.message.max.bytes</li> - * <li>auto.offset.reset with the values "largest", "smallest"</li> - * <li>fetch.wait.max.ms</li> - * </ul> - * </li> - * </ul> - * - * <h1>Offset handling</h1> - * - * <p>Offsets whose records have been read and are checkpointed will be committed back to ZooKeeper - * by the offset handler. In addition, the offset handler finds the point where the source initially - * starts reading from the stream, when the streaming job is started.</p> - * - * <p>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.</p> - * - * <p>If checkpointing is disabled, the consumer will periodically commit the current offset - * to Zookeeper.</p> - * - * <p>When using a Kafka topic to send data between Flink jobs, we recommend using the - * {@see TypeInformationSerializationSchema} and {@see TypeInformationKeyValueSerializationSchema}.</p> - * - * <p><b>NOTE:</b> 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.</p> - */ -public class FlinkKafkaConsumer08<T> extends FlinkKafkaConsumerBase<T> { - - private static final long serialVersionUID = -6272159445203409112L; - - /** Configuration key for the number of retries for getting the partition info */ - public static final String GET_PARTITIONS_RETRIES_KEY = "flink.get-partitions.retry"; - - /** Default number of retries for getting the partition info. One retry means going through the full list of brokers */ - public static final int DEFAULT_GET_PARTITIONS_RETRIES = 3; - - // ------------------------------------------------------------------------ - - /** The properties to parametrize the Kafka consumer and ZooKeeper client */ - private final Properties kafkaProperties; - - /** The behavior when encountering an invalid offset (see {@link OffsetRequest}) */ - private final long invalidOffsetBehavior; - - /** The interval in which to automatically commit (-1 if deactivated) */ - private final long autoCommitInterval; - - // ------------------------------------------------------------------------ - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.8.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 FlinkKafkaConsumer08(String topic, DeserializationSchema<T> valueDeserializer, Properties props) { - this(Collections.singletonList(topic), valueDeserializer, props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.8.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 FlinkKafkaConsumer08(String topic, KeyedDeserializationSchema<T> deserializer, Properties props) { - this(Collections.singletonList(topic), deserializer, props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.8.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 FlinkKafkaConsumer08(List<String> topics, DeserializationSchema<T> deserializer, Properties props) { - this(topics, new KeyedDeserializationSchemaWrapper<>(deserializer), props); - } - - /** - * Creates a new Kafka streaming source consumer for Kafka 0.8.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 FlinkKafkaConsumer08(List<String> topics, KeyedDeserializationSchema<T> deserializer, Properties props) { - super(topics, deserializer); - - checkNotNull(topics, "topics"); - this.kafkaProperties = checkNotNull(props, "props"); - - // validate the zookeeper properties - validateZooKeeperConfig(props); - - this.invalidOffsetBehavior = getInvalidOffsetBehavior(props); - this.autoCommitInterval = PropertiesUtil.getLong(props, "auto.commit.interval.ms", 60000); - } - - @Override - protected AbstractFetcher<T, ?> createFetcher( - SourceContext<T> sourceContext, - List<KafkaTopicPartition> thisSubtaskPartitions, - SerializedValue<AssignerWithPeriodicWatermarks<T>> watermarksPeriodic, - SerializedValue<AssignerWithPunctuatedWatermarks<T>> watermarksPunctuated, - StreamingRuntimeContext runtimeContext) throws Exception { - - boolean useMetrics = !Boolean.valueOf(kafkaProperties.getProperty(KEY_DISABLE_METRICS, "false")); - - return new Kafka08Fetcher<>(sourceContext, thisSubtaskPartitions, - watermarksPeriodic, watermarksPunctuated, - runtimeContext, deserializer, kafkaProperties, - invalidOffsetBehavior, autoCommitInterval, useMetrics); - } - - @Override - protected List<KafkaTopicPartition> getKafkaPartitions(List<String> topics) { - // Connect to a broker to get the partitions for all topics - List<KafkaTopicPartition> partitionInfos = - KafkaTopicPartition.dropLeaderData(getPartitionsForTopic(topics, kafkaProperties)); - - if (partitionInfos.size() == 0) { - throw new RuntimeException( - "Unable to retrieve any partitions for the requested topics " + topics + - ". Please check previous log entries"); - } - - if (LOG.isInfoEnabled()) { - logPartitionInfo(LOG, partitionInfos); - } - - return partitionInfos; - } - - // ------------------------------------------------------------------------ - // Kafka / ZooKeeper communication utilities - // ------------------------------------------------------------------------ - - /** - * Send request to Kafka to get partitions for topic. - * - * @param topics The name of the topics. - * @param properties The properties for the Kafka Consumer that is used to query the partitions for the topic. - */ - public static List<KafkaTopicPartitionLeader> getPartitionsForTopic(List<String> topics, Properties properties) { - String seedBrokersConfString = properties.getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG); - final int numRetries = getInt(properties, GET_PARTITIONS_RETRIES_KEY, DEFAULT_GET_PARTITIONS_RETRIES); - - checkNotNull(seedBrokersConfString, "Configuration property %s not set", ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG); - String[] seedBrokers = seedBrokersConfString.split(","); - List<KafkaTopicPartitionLeader> partitions = new ArrayList<>(); - - final String clientId = "flink-kafka-consumer-partition-lookup"; - final int soTimeout = getInt(properties, "socket.timeout.ms", 30000); - final int bufferSize = getInt(properties, "socket.receive.buffer.bytes", 65536); - - Random rnd = new Random(); - retryLoop: for (int retry = 0; retry < numRetries; retry++) { - // we pick a seed broker randomly to avoid overloading the first broker with all the requests when the - // parallel source instances start. Still, we try all available brokers. - int index = rnd.nextInt(seedBrokers.length); - brokersLoop: for (int arrIdx = 0; arrIdx < seedBrokers.length; arrIdx++) { - String seedBroker = seedBrokers[index]; - LOG.info("Trying to get topic metadata from broker {} in try {}/{}", seedBroker, retry, numRetries); - if (++index == seedBrokers.length) { - index = 0; - } - - URL brokerUrl = NetUtils.getCorrectHostnamePort(seedBroker); - SimpleConsumer consumer = null; - try { - consumer = new SimpleConsumer(brokerUrl.getHost(), brokerUrl.getPort(), soTimeout, bufferSize, clientId); - - TopicMetadataRequest req = new TopicMetadataRequest(topics); - kafka.javaapi.TopicMetadataResponse resp = consumer.send(req); - - List<TopicMetadata> metaData = resp.topicsMetadata(); - - // clear in case we have an incomplete list from previous tries - partitions.clear(); - for (TopicMetadata item : metaData) { - if (item.errorCode() != ErrorMapping.NoError()) { - // warn and try more brokers - LOG.warn("Error while getting metadata from broker " + seedBroker + " to find partitions " + - "for " + topics.toString() + ". Error: " + ErrorMapping.exceptionFor(item.errorCode()).getMessage()); - continue brokersLoop; - } - if (!topics.contains(item.topic())) { - LOG.warn("Received metadata from topic " + item.topic() + " even though it was not requested. Skipping ..."); - continue brokersLoop; - } - for (PartitionMetadata part : item.partitionsMetadata()) { - Node leader = brokerToNode(part.leader()); - KafkaTopicPartition ktp = new KafkaTopicPartition(item.topic(), part.partitionId()); - KafkaTopicPartitionLeader pInfo = new KafkaTopicPartitionLeader(ktp, leader); - partitions.add(pInfo); - } - } - break retryLoop; // leave the loop through the brokers - } catch (Exception e) { - //validates seed brokers in case of a ClosedChannelException - validateSeedBrokers(seedBrokers, e); - LOG.warn("Error communicating with broker " + seedBroker + " to find partitions for " + topics.toString() + "." + - "" + e.getClass() + ". Message: " + e.getMessage()); - LOG.debug("Detailed trace", e); - // we sleep a bit. Retrying immediately doesn't make sense in cases where Kafka is reorganizing the leader metadata - try { - Thread.sleep(500); - } catch (InterruptedException e1) { - // sleep shorter. - } - } finally { - if (consumer != null) { - consumer.close(); - } - } - } // brokers loop - } // retries loop - return partitions; - } - - /** - * Turn a broker instance into a node instance - * @param broker broker instance - * @return Node representing the given broker - */ - private static Node brokerToNode(Broker broker) { - return new Node(broker.id(), broker.host(), broker.port()); - } - - /** - * Validate the ZK configuration, checking for required parameters - * @param props Properties to check - */ - protected static void validateZooKeeperConfig(Properties props) { - if (props.getProperty("zookeeper.connect") == null) { - throw new IllegalArgumentException("Required property 'zookeeper.connect' has not been set in the properties"); - } - if (props.getProperty(ConsumerConfig.GROUP_ID_CONFIG) == null) { - throw new IllegalArgumentException("Required property '" + ConsumerConfig.GROUP_ID_CONFIG - + "' has not been set in the properties"); - } - - try { - //noinspection ResultOfMethodCallIgnored - Integer.parseInt(props.getProperty("zookeeper.session.timeout.ms", "0")); - } - catch (NumberFormatException e) { - throw new IllegalArgumentException("Property 'zookeeper.session.timeout.ms' is not a valid integer"); - } - - try { - //noinspection ResultOfMethodCallIgnored - Integer.parseInt(props.getProperty("zookeeper.connection.timeout.ms", "0")); - } - catch (NumberFormatException e) { - throw new IllegalArgumentException("Property 'zookeeper.connection.timeout.ms' is not a valid integer"); - } - } - - /** - * Validate that at least one seed broker is valid in case of a - * ClosedChannelException. - * - * @param seedBrokers - * array containing the seed brokers e.g. ["host1:port1", - * "host2:port2"] - * @param exception - * instance - */ - private static void validateSeedBrokers(String[] seedBrokers, Exception exception) { - if (!(exception instanceof ClosedChannelException)) { - return; - } - int unknownHosts = 0; - for (String broker : seedBrokers) { - URL brokerUrl = NetUtils.getCorrectHostnamePort(broker.trim()); - try { - InetAddress.getByName(brokerUrl.getHost()); - } catch (UnknownHostException e) { - unknownHosts++; - } - } - // throw meaningful exception if all the provided hosts are invalid - if (unknownHosts == seedBrokers.length) { - throw new IllegalArgumentException("All the servers provided in: '" - + ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + "' config are invalid. (unknown hosts)"); - } - } - - private static long getInvalidOffsetBehavior(Properties config) { - final String val = config.getProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "largest"); - if (val.equals("none")) { - throw new IllegalArgumentException("Cannot use '" + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG - + "' value 'none'. Possible values: 'latest', 'largest', or 'earliest'."); - } - else if (val.equals("largest") || val.equals("latest")) { // largest is kafka 0.8, latest is kafka 0.9 - return OffsetRequest.LatestTime(); - } else { - return OffsetRequest.EarliestTime(); - } - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer081.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer081.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer081.java deleted file mode 100644 index 56ccd0b..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer081.java +++ /dev/null @@ -1,39 +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.util.serialization.DeserializationSchema; - -import java.util.Properties; - -/** - * THIS CLASS IS DEPRECATED. Use FlinkKafkaConsumer08 instead. - */ -@Deprecated -public class FlinkKafkaConsumer081<T> extends FlinkKafkaConsumer08<T> { - - private static final long serialVersionUID = -5649906773771949146L; - - /** - * THIS CONSTRUCTOR IS DEPRECATED. Use FlinkKafkaConsumer08 instead. - */ - @Deprecated - public FlinkKafkaConsumer081(String topic, DeserializationSchema<T> valueDeserializer, Properties props) { - super(topic, valueDeserializer, props); - } -} - http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082.java deleted file mode 100644 index 0520336..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaConsumer082.java +++ /dev/null @@ -1,39 +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.util.serialization.DeserializationSchema; - -import java.util.Properties; - -/** - * THIS CLASS IS DEPRECATED. Use FlinkKafkaConsumer08 instead. - */ -@Deprecated -public class FlinkKafkaConsumer082<T> extends FlinkKafkaConsumer08<T> { - - private static final long serialVersionUID = -5649906773771949146L; - - /** - * THIS CONSTRUCTOR IS DEPRECATED. Use FlinkKafkaConsumer08 instead. - */ - @Deprecated - public FlinkKafkaConsumer082(String topic, DeserializationSchema<T> valueDeserializer, Properties props) { - super(topic, valueDeserializer, props); - } -} - http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java deleted file mode 100644 index 1c2e0b7..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer.java +++ /dev/null @@ -1,64 +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.connectors.kafka.partitioner.KafkaPartitioner; -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 java.util.Properties; - - -/** - * THIS CLASS IS DEPRECATED. Use FlinkKafkaProducer08 instead. - */ -@Deprecated -public class FlinkKafkaProducer<IN> extends FlinkKafkaProducer08<IN> { - - @Deprecated - public FlinkKafkaProducer(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) { - super(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), null); - } - - @Deprecated - public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) { - super(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, null); - } - - @Deprecated - public FlinkKafkaProducer(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner customPartitioner) { - super(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner); - - } - - @Deprecated - public FlinkKafkaProducer(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) { - super(topicId, serializationSchema, getPropertiesFromBrokerList(brokerList), null); - } - - @Deprecated - public FlinkKafkaProducer(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) { - super(topicId, serializationSchema, producerConfig, null); - } - - @Deprecated - public FlinkKafkaProducer(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner customPartitioner) { - super(topicId, serializationSchema, producerConfig, customPartitioner); - } - -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java deleted file mode 100644 index 65de5fc..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/FlinkKafkaProducer08.java +++ /dev/null @@ -1,145 +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.connectors.kafka.partitioner.FixedPartitioner; -import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner; -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 java.util.Properties; - - -/** - * Flink Sink to produce data into a Kafka topic. This producer is compatible with Kafka 0.8. - * - * Please note that this producer does not have any reliability guarantees. - * - * @param <IN> Type of the messages to write into Kafka. - */ -public class FlinkKafkaProducer08<IN> extends FlinkKafkaProducerBase<IN> { - - private static final long serialVersionUID = 1L; - - // ------------------- Keyless serialization schema constructors ---------------------- - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to - * the topic. - * - * @param brokerList - * Comma separated addresses of the brokers - * @param topicId - * ID of the Kafka topic. - * @param serializationSchema - * User defined (keyless) serialization schema. - */ - public FlinkKafkaProducer08(String brokerList, String topicId, SerializationSchema<IN> serializationSchema) { - this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>()); - } - - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to - * the topic. - * - * @param topicId - * ID of the Kafka topic. - * @param serializationSchema - * User defined (keyless) serialization schema. - * @param producerConfig - * Properties with the producer configuration. - */ - public FlinkKafkaProducer08(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig) { - this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, new FixedPartitioner<IN>()); - } - - /** - * The main constructor for creating a FlinkKafkaProducer. - * - * @param topicId The topic to write data to - * @param serializationSchema A (keyless) serializable serialization schema for turning user objects into a kafka-consumable byte[] - * @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 FlinkKafkaProducer08(String topicId, SerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) { - this(topicId, new KeyedSerializationSchemaWrapper<>(serializationSchema), producerConfig, customPartitioner); - - } - - // ------------------- Key/Value serialization schema constructors ---------------------- - - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to - * the topic. - * - * @param brokerList - * Comma separated addresses of the brokers - * @param topicId - * ID of the Kafka topic. - * @param serializationSchema - * User defined serialization schema supporting key/value messages - */ - public FlinkKafkaProducer08(String brokerList, String topicId, KeyedSerializationSchema<IN> serializationSchema) { - this(topicId, serializationSchema, getPropertiesFromBrokerList(brokerList), new FixedPartitioner<IN>()); - } - - /** - * Creates a FlinkKafkaProducer for a given topic. The sink produces its input to - * the topic. - * - * @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 FlinkKafkaProducer08(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig) { - this(topicId, serializationSchema, producerConfig, new FixedPartitioner<IN>()); - } - - /** - * The main constructor for creating a FlinkKafkaProducer. - * - * @param topicId The topic to write data to - * @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 assining messages to Kafka partitions. - */ - public FlinkKafkaProducer08(String topicId, KeyedSerializationSchema<IN> serializationSchema, Properties producerConfig, KafkaPartitioner<IN> customPartitioner) { - super(topicId, serializationSchema, producerConfig, customPartitioner); - } - - @Override - protected void flush() { - // The Kafka 0.8 producer doesn't support flushing, we wait here - // until all pending records are confirmed - synchronized (pendingRecordsLock) { - while (pendingRecords > 0) { - try { - pendingRecordsLock.wait(); - } catch (InterruptedException e) { - // this can be interrupted when the Task has been cancelled. - // by throwing an exception, we ensure that this checkpoint doesn't get confirmed - throw new RuntimeException("Flushing got interrupted while checkpointing", e); - } - } - } - } - -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.java deleted file mode 100644 index b155576..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSink.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.connectors.kafka; - -import org.apache.flink.api.table.Row; -import org.apache.flink.streaming.connectors.kafka.partitioner.KafkaPartitioner; -import org.apache.flink.streaming.util.serialization.SerializationSchema; - -import java.util.Properties; - -/** - * Kafka 0.8 {@link KafkaTableSink} that serializes data in JSON format. - */ -public class Kafka08JsonTableSink extends KafkaJsonTableSink { - - /** - * Creates {@link KafkaTableSink} for Kafka 0.8 - * - * @param topic topic in Kafka to which table is written - * @param properties properties to connect to Kafka - * @param partitioner Kafka partitioner - */ - public Kafka08JsonTableSink(String topic, Properties properties, KafkaPartitioner<Row> partitioner) { - super(topic, properties, partitioner); - } - - @Override - protected FlinkKafkaProducerBase<Row> createKafkaProducer(String topic, Properties properties, SerializationSchema<Row> serializationSchema, KafkaPartitioner<Row> partitioner) { - return new FlinkKafkaProducer08<>(topic, serializationSchema, properties, partitioner); - } - - @Override - protected Kafka08JsonTableSink createCopy() { - return new Kafka08JsonTableSink(topic, properties, partitioner); - } -} - http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java deleted file mode 100644 index 63bb57e..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08JsonTableSource.java +++ /dev/null @@ -1,71 +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.typeinfo.TypeInformation; -import org.apache.flink.api.table.Row; -import org.apache.flink.api.table.sources.StreamTableSource; -import org.apache.flink.streaming.util.serialization.DeserializationSchema; - -import java.util.Properties; - -/** - * Kafka {@link StreamTableSource} for Kafka 0.8. - */ -public class Kafka08JsonTableSource extends KafkaJsonTableSource { - - /** - * Creates a Kafka 0.8 JSON {@link StreamTableSource}. - * - * @param topic Kafka topic to consume. - * @param properties Properties for the Kafka consumer. - * @param fieldNames Row field names. - * @param fieldTypes Row field types. - */ - public Kafka08JsonTableSource( - String topic, - Properties properties, - String[] fieldNames, - TypeInformation<?>[] fieldTypes) { - - super(topic, properties, fieldNames, fieldTypes); - } - - /** - * Creates a Kafka 0.8 JSON {@link StreamTableSource}. - * - * @param topic Kafka topic to consume. - * @param properties Properties for the Kafka consumer. - * @param fieldNames Row field names. - * @param fieldTypes Row field types. - */ - public Kafka08JsonTableSource( - String topic, - Properties properties, - String[] fieldNames, - Class<?>[] fieldTypes) { - - super(topic, properties, fieldNames, fieldTypes); - } - - @Override - FlinkKafkaConsumerBase<Row> getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) { - return new FlinkKafkaConsumer08<>(topic, deserializationSchema, properties); - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java deleted file mode 100644 index 8f51237..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka08TableSource.java +++ /dev/null @@ -1,75 +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.typeinfo.TypeInformation; -import org.apache.flink.api.table.Row; -import org.apache.flink.api.table.sources.StreamTableSource; -import org.apache.flink.streaming.util.serialization.DeserializationSchema; - -import java.util.Properties; - -/** - * Kafka {@link StreamTableSource} for Kafka 0.8. - */ -public class Kafka08TableSource extends KafkaTableSource { - - /** - * Creates a Kafka 0.8 {@link StreamTableSource}. - * - * @param topic Kafka topic to consume. - * @param properties Properties for the Kafka consumer. - * @param deserializationSchema Deserialization schema to use for Kafka records. - * @param fieldNames Row field names. - * @param fieldTypes Row field types. - */ - public Kafka08TableSource( - String topic, - Properties properties, - DeserializationSchema<Row> deserializationSchema, - String[] fieldNames, - TypeInformation<?>[] fieldTypes) { - - super(topic, properties, deserializationSchema, fieldNames, fieldTypes); - } - - /** - * Creates a Kafka 0.8 {@link StreamTableSource}. - * - * @param topic Kafka topic to consume. - * @param properties Properties for the Kafka consumer. - * @param deserializationSchema Deserialization schema to use for Kafka records. - * @param fieldNames Row field names. - * @param fieldTypes Row field types. - */ - public Kafka08TableSource( - String topic, - Properties properties, - DeserializationSchema<Row> deserializationSchema, - String[] fieldNames, - Class<?>[] fieldTypes) { - - super(topic, properties, deserializationSchema, fieldNames, fieldTypes); - } - - @Override - FlinkKafkaConsumerBase<Row> getKafkaConsumer(String topic, Properties properties, DeserializationSchema<Row> deserializationSchema) { - return new FlinkKafkaConsumer08<>(topic, deserializationSchema, properties); - } -} http://git-wip-us.apache.org/repos/asf/flink/blob/de4fe3b7/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java ---------------------------------------------------------------------- diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java deleted file mode 100644 index 23ff276..0000000 --- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/ClosableBlockingQueue.java +++ /dev/null @@ -1,507 +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.internals; - -import java.util.ArrayDeque; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.Iterator; -import java.util.List; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.ReentrantLock; - -import static java.util.Objects.requireNonNull; - -/** - * A special form of blocking queue with two additions: - * <ol> - * <li>The queue can be closed atomically when empty. Adding elements after the queue - * is closed fails. This allows queue consumers to atomically discover that no elements - * are available and mark themselves as shut down.</li> - * <li>The queue allows to poll batches of elements in one polling call.</li> - * </ol> - * - * The queue has no capacity restriction and is safe for multiple producers and consumers. - * - * <p>Note: Null elements are prohibited. - * - * @param <E> The type of elements in the queue. - */ -public class ClosableBlockingQueue<E> { - - /** The lock used to make queue accesses and open checks atomic */ - private final ReentrantLock lock; - - /** The condition on which blocking get-calls wait if the queue is empty */ - private final Condition nonEmpty; - - /** The deque of elements */ - private final ArrayDeque<E> elements; - - /** Flag marking the status of the queue */ - private volatile boolean open; - - // ------------------------------------------------------------------------ - - /** - * Creates a new empty queue. - */ - public ClosableBlockingQueue() { - this(10); - } - - /** - * Creates a new empty queue, reserving space for at least the specified number - * of elements. The queu can still grow, of more elements are added than the - * reserved space. - * - * @param initialSize The number of elements to reserve space for. - */ - public ClosableBlockingQueue(int initialSize) { - this.lock = new ReentrantLock(true); - this.nonEmpty = this.lock.newCondition(); - - this.elements = new ArrayDeque<>(initialSize); - this.open = true; - - - } - - /** - * Creates a new queue that contains the given elements. - * - * @param initialElements The elements to initially add to the queue. - */ - public ClosableBlockingQueue(Collection<? extends E> initialElements) { - this(initialElements.size()); - this.elements.addAll(initialElements); - } - - // ------------------------------------------------------------------------ - // Size and status - // ------------------------------------------------------------------------ - - /** - * Gets the number of elements currently in the queue. - * @return The number of elements currently in the queue. - */ - public int size() { - lock.lock(); - try { - return elements.size(); - } finally { - lock.unlock(); - } - } - - /** - * Checks whether the queue is empty (has no elements). - * @return True, if the queue is empty; false, if it is non-empty. - */ - public boolean isEmpty() { - return size() == 0; - } - - /** - * Checks whether the queue is currently open, meaning elements can be added and polled. - * @return True, if the queue is open; false, if it is closed. - */ - public boolean isOpen() { - return open; - } - - /** - * Tries to close the queue. Closing the queue only succeeds when no elements are - * in the queue when this method is called. Checking whether the queue is empty, and - * marking the queue as closed is one atomic operation. - * - * @return True, if the queue is closed, false if the queue remains open. - */ - public boolean close() { - lock.lock(); - try { - if (open) { - if (elements.isEmpty()) { - open = false; - nonEmpty.signalAll(); - return true; - } else { - return false; - } - } - else { - // already closed - return true; - } - } finally { - lock.unlock(); - } - } - - // ------------------------------------------------------------------------ - // Adding / Removing elements - // ------------------------------------------------------------------------ - - /** - * Tries to add an element to the queue, if the queue is still open. Checking whether the queue - * is open and adding the element is one atomic operation. - * - * <p>Unlike the {@link #add(Object)} method, this method never throws an exception, - * but only indicates via the return code if the element was added or the - * queue was closed. - * - * @param element The element to add. - * @return True, if the element was added, false if the queue was closes. - */ - public boolean addIfOpen(E element) { - requireNonNull(element); - - lock.lock(); - try { - if (open) { - elements.addLast(element); - if (elements.size() == 1) { - nonEmpty.signalAll(); - } - } - return open; - } finally { - lock.unlock(); - } - } - - /** - * Adds the element to the queue, or fails with an exception, if the queue is closed. - * Checking whether the queue is open and adding the element is one atomic operation. - * - * @param element The element to add. - * @throws IllegalStateException Thrown, if the queue is closed. - */ - public void add(E element) throws IllegalStateException { - requireNonNull(element); - - lock.lock(); - try { - if (open) { - elements.addLast(element); - if (elements.size() == 1) { - nonEmpty.signalAll(); - } - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Returns the queue's next element without removing it, if the queue is non-empty. - * Otherwise, returns null. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and getting the next element is one atomic operation. - * - * <p>This method never blocks. - * - * @return The queue's next element, or null, if the queue is empty. - * @throws IllegalStateException Thrown, if the queue is closed. - */ - public E peek() { - lock.lock(); - try { - if (open) { - if (elements.size() > 0) { - return elements.getFirst(); - } else { - return null; - } - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Returns the queue's next element and removes it, the queue is non-empty. - * Otherwise, this method returns null. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the next element is one atomic operation. - * - * <p>This method never blocks. - * - * @return The queue's next element, or null, if the queue is empty. - * @throws IllegalStateException Thrown, if the queue is closed. - */ - public E poll() { - lock.lock(); - try { - if (open) { - if (elements.size() > 0) { - return elements.removeFirst(); - } else { - return null; - } - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Returns all of the queue's current elements in a list, if the queue is non-empty. - * Otherwise, this method returns null. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the elements is one atomic operation. - * - * <p>This method never blocks. - * - * @return All of the queue's elements, or null, if the queue is empty. - * @throws IllegalStateException Thrown, if the queue is closed. - */ - public List<E> pollBatch() { - lock.lock(); - try { - if (open) { - if (elements.size() > 0) { - ArrayList<E> result = new ArrayList<>(elements); - elements.clear(); - return result; - } else { - return null; - } - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Returns the next element in the queue. If the queue is empty, this method - * waits until at least one element is added. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the next element is one atomic operation. - * - * @return The next element in the queue, never null. - * - * @throws IllegalStateException Thrown, if the queue is closed. - * @throws InterruptedException Throw, if the thread is interrupted while waiting for an - * element to be added. - */ - public E getElementBlocking() throws InterruptedException { - lock.lock(); - try { - while (open && elements.isEmpty()) { - nonEmpty.await(); - } - - if (open) { - return elements.removeFirst(); - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Returns the next element in the queue. If the queue is empty, this method - * waits at most a certain time until an element becomes available. If no element - * is available after that time, the method returns null. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the next element is one atomic operation. - * - * @param timeoutMillis The number of milliseconds to block, at most. - * @return The next element in the queue, or null, if the timeout expires before an element is available. - * - * @throws IllegalStateException Thrown, if the queue is closed. - * @throws InterruptedException Throw, if the thread is interrupted while waiting for an - * element to be added. - */ - public E getElementBlocking(long timeoutMillis) throws InterruptedException { - if (timeoutMillis == 0L) { - // wait forever case - return getElementBlocking(); - } else if (timeoutMillis < 0L) { - throw new IllegalArgumentException("invalid timeout"); - } - - final long deadline = System.currentTimeMillis() + timeoutMillis; - - lock.lock(); - try { - while (open && elements.isEmpty() && timeoutMillis > 0) { - nonEmpty.await(timeoutMillis, TimeUnit.MILLISECONDS); - timeoutMillis = deadline - System.currentTimeMillis(); - } - - if (!open) { - throw new IllegalStateException("queue is closed"); - } - else if (elements.isEmpty()) { - return null; - } else { - return elements.removeFirst(); - } - } finally { - lock.unlock(); - } - } - - /** - * Gets all the elements found in the list, or blocks until at least one element - * was added. If the queue is empty when this method is called, it blocks until - * at least one element is added. - * - * <p>This method always returns a list with at least one element. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the next element is one atomic operation. - * - * @return A list with all elements in the queue, always at least one element. - * - * @throws IllegalStateException Thrown, if the queue is closed. - * @throws InterruptedException Throw, if the thread is interrupted while waiting for an - * element to be added. - */ - public List<E> getBatchBlocking() throws InterruptedException { - lock.lock(); - try { - while (open && elements.isEmpty()) { - nonEmpty.await(); - } - if (open) { - ArrayList<E> result = new ArrayList<>(elements); - elements.clear(); - return result; - } else { - throw new IllegalStateException("queue is closed"); - } - } finally { - lock.unlock(); - } - } - - /** - * Gets all the elements found in the list, or blocks until at least one element - * was added. This method is similar as {@link #getBatchBlocking()}, but takes - * a number of milliseconds that the method will maximally wait before returning. - * - * <p>This method never returns null, but an empty list, if the queue is empty when - * the method is called and the request times out before an element was added. - * - * <p>The method throws an {@code IllegalStateException} if the queue is closed. - * Checking whether the queue is open and removing the next element is one atomic operation. - * - * @param timeoutMillis The number of milliseconds to wait, at most. - * @return A list with all elements in the queue, possible an empty list. - * - * @throws IllegalStateException Thrown, if the queue is closed. - * @throws InterruptedException Throw, if the thread is interrupted while waiting for an - * element to be added. - */ - public List<E> getBatchBlocking(long timeoutMillis) throws InterruptedException { - if (timeoutMillis == 0L) { - // wait forever case - return getBatchBlocking(); - } else if (timeoutMillis < 0L) { - throw new IllegalArgumentException("invalid timeout"); - } - - final long deadline = System.currentTimeMillis() + timeoutMillis; - - lock.lock(); - try { - while (open && elements.isEmpty() && timeoutMillis > 0) { - nonEmpty.await(timeoutMillis, TimeUnit.MILLISECONDS); - timeoutMillis = deadline - System.currentTimeMillis(); - } - - if (!open) { - throw new IllegalStateException("queue is closed"); - } - else if (elements.isEmpty()) { - return Collections.emptyList(); - } - else { - ArrayList<E> result = new ArrayList<>(elements); - elements.clear(); - return result; - } - } finally { - lock.unlock(); - } - } - - // ------------------------------------------------------------------------ - // Standard Utilities - // ------------------------------------------------------------------------ - - @Override - public int hashCode() { - int hashCode = 17; - for (E element : elements) { - hashCode = 31 * hashCode + element.hashCode(); - } - return hashCode; - } - - @Override - public boolean equals(Object obj) { - if (obj == this) { - return true; - } else if (obj != null && obj.getClass() == ClosableBlockingQueue.class) { - @SuppressWarnings("unchecked") - ClosableBlockingQueue<E> that = (ClosableBlockingQueue<E>) obj; - - if (this.elements.size() == that.elements.size()) { - Iterator<E> thisElements = this.elements.iterator(); - for (E thatNext : that.elements) { - E thisNext = thisElements.next(); - if (!(thisNext == null ? thatNext == null : thisNext.equals(thatNext))) { - return false; - } - } - return true; - } else { - return false; - } - } else { - return false; - } - } - - @Override - public String toString() { - return elements.toString(); - } -}
