[ https://issues.apache.org/jira/browse/FLINK-2098?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14567903#comment-14567903 ]
ASF GitHub Bot commented on FLINK-2098: --------------------------------------- Github user aljoscha commented on a diff in the pull request: https://github.com/apache/flink/pull/742#discussion_r31463699 --- Diff: flink-staging/flink-streaming/flink-streaming-connectors/flink-connector-twitter/src/main/java/org/apache/flink/streaming/connectors/twitter/TwitterSource.java --- @@ -1,322 +1,322 @@ -/* - * 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.twitter; - -import java.io.FileInputStream; -import java.io.InputStream; -import java.util.Properties; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; - -import org.apache.flink.configuration.Configuration; -import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction; -import org.apache.flink.streaming.api.functions.source.SourceFunction; -import org.apache.flink.util.Collector; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.twitter.hbc.ClientBuilder; -import com.twitter.hbc.core.Constants; -import com.twitter.hbc.core.endpoint.StatusesSampleEndpoint; -import com.twitter.hbc.core.processor.StringDelimitedProcessor; -import com.twitter.hbc.httpclient.BasicClient; -import com.twitter.hbc.httpclient.auth.Authentication; -import com.twitter.hbc.httpclient.auth.OAuth1; - -/** - * Implementation of {@link SourceFunction} specialized to emit tweets from - * Twitter. It can connect to Twitter Streaming API, collect tweets and - */ -public class TwitterSource extends RichParallelSourceFunction<String> { - - private static final Logger LOG = LoggerFactory.getLogger(TwitterSource.class); - - private static final long serialVersionUID = 1L; - private String authPath; - private transient BlockingQueue<String> queue; - private int queueSize = 10000; - private transient BasicClient client; - private int waitSec = 5; - - private int maxNumberOfTweets; - private int currentNumberOfTweets; - - private String nextElement = null; - - private volatile boolean isRunning = false; - - /** - * Create {@link TwitterSource} for streaming - * - * @param authPath - * Location of the properties file containing the required - * authentication information. - */ - public TwitterSource(String authPath) { - this.authPath = authPath; - maxNumberOfTweets = -1; - } - - /** - * Create {@link TwitterSource} to collect finite number of tweets - * - * @param authPath - * Location of the properties file containing the required - * authentication information. - * @param numberOfTweets - * - */ - public TwitterSource(String authPath, int numberOfTweets) { - this.authPath = authPath; - this.maxNumberOfTweets = numberOfTweets; - } - - @Override - public void open(Configuration parameters) throws Exception { - initializeConnection(); - currentNumberOfTweets = 0; - } - - /** - * Initialize Hosebird Client to be able to consume Twitter's Streaming API - */ - private void initializeConnection() { - - if (LOG.isInfoEnabled()) { - LOG.info("Initializing Twitter Streaming API connection"); - } - - queue = new LinkedBlockingQueue<String>(queueSize); - - StatusesSampleEndpoint endpoint = new StatusesSampleEndpoint(); - endpoint.stallWarnings(false); - - Authentication auth = authenticate(); - - initializeClient(endpoint, auth); - - if (LOG.isInfoEnabled()) { - LOG.info("Twitter Streaming API connection established successfully"); - } - } - - private OAuth1 authenticate() { - - Properties authenticationProperties = loadAuthenticationProperties(); - - return new OAuth1(authenticationProperties.getProperty("consumerKey"), - authenticationProperties.getProperty("consumerSecret"), - authenticationProperties.getProperty("token"), - authenticationProperties.getProperty("secret")); - } - - /** - * Reads the given properties file for the authentication data. - * - * @return the authentication data. - */ - private Properties loadAuthenticationProperties() { - Properties properties = new Properties(); - try { - InputStream input = new FileInputStream(authPath); - properties.load(input); - input.close(); - } catch (Exception e) { - throw new RuntimeException("Cannot open .properties file: " + authPath, e); - } - return properties; - } - - private void initializeClient(StatusesSampleEndpoint endpoint, Authentication auth) { - - client = new ClientBuilder().name("twitterSourceClient").hosts(Constants.STREAM_HOST) - .endpoint(endpoint).authentication(auth) - .processor(new StringDelimitedProcessor(queue)).build(); - - client.connect(); - } - - /** - * Put tweets into output - * - * @param collector - * Collector in which the tweets are collected. - */ - protected void collectFiniteMessages(Collector<String> collector) { - - if (LOG.isInfoEnabled()) { - LOG.info("Collecting tweets"); - } - - for (int i = 0; i < maxNumberOfTweets; i++) { - collectOneMessage(collector); - } - - if (LOG.isInfoEnabled()) { - LOG.info("Collecting tweets finished"); - } - } - - /** - * Put tweets into output - * - * @param collector - * Collector in which the tweets are collected. - */ - protected void collectMessages(Collector<String> collector) { - - if (LOG.isInfoEnabled()) { - LOG.info("Tweet-stream begins"); - } - - while (isRunning) { - collectOneMessage(collector); - } - } - - /** - * Put one tweet into the output. - * - * @param collector - * Collector in which the tweets are collected. - */ - protected void collectOneMessage(Collector<String> collector) { - if (client.isDone()) { - if (LOG.isErrorEnabled()) { - LOG.error("Client connection closed unexpectedly: {}", client.getExitEvent() - .getMessage()); - } - } - - try { - String msg = queue.poll(waitSec, TimeUnit.SECONDS); - if (msg != null) { - collector.collect(msg); - } else { - if (LOG.isInfoEnabled()) { - LOG.info("Did not receive a message in {} seconds", waitSec); - } - } - } catch (InterruptedException e) { - throw new RuntimeException("'Waiting for tweet' thread is interrupted", e); - } - - } - - private void closeConnection() { - - if (LOG.isInfoEnabled()) { - LOG.info("Initiating connection close"); - } - - client.stop(); - - if (LOG.isInfoEnabled()) { - LOG.info("Connection closed successfully"); - } - } - - /** - * Get the size of the queue in which the tweets are contained temporarily. - * - * @return the size of the queue in which the tweets are contained - * temporarily - */ - public int getQueueSize() { - return queueSize; - } - - /** - * Set the size of the queue in which the tweets are contained temporarily. - * - * @param queueSize - * The desired value. - */ - public void setQueueSize(int queueSize) { - this.queueSize = queueSize; - } - - /** - * This function tells how long TwitterSource waits for the tweets. - * - * @return Number of second. - */ - public int getWaitSec() { - return waitSec; - } - - /** - * This function sets how long TwitterSource should wait for the tweets. - * - * @param waitSec - * The desired value. - */ - public void setWaitSec(int waitSec) { - this.waitSec = waitSec; - } - - @Override - public boolean reachedEnd() throws Exception { - if (currentNumberOfTweets >= maxNumberOfTweets) { - return false; - } - - if (nextElement != null) { - return true; - } - if (client.isDone()) { - if (LOG.isErrorEnabled()) { - LOG.error("Client connection closed unexpectedly: {}", client.getExitEvent() - .getMessage()); - } - return false; - } - - try { - String msg = queue.poll(waitSec, TimeUnit.SECONDS); - if (msg != null) { - nextElement = msg; - return true; - } else { - if (LOG.isInfoEnabled()) { - LOG.info("Did not receive a message in {} seconds", waitSec); - } - } - } catch (InterruptedException e) { - throw new RuntimeException("'Waiting for tweet' thread is interrupted", e); - } - return false; - } - - @Override - public String next() throws Exception { - if (nextElement != null) { - String result = nextElement; - nextElement = null; - return result; - } - if (reachedEnd()) { - throw new RuntimeException("Twitter stream end reached."); - } else { - String result = nextElement; - nextElement = null; - return result; - } - } -} +///* +// * 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. --- End diff -- This is because I'm waiting since the recent TwitterSource PR is in. > Checkpoint barrier initiation at source is not aligned with snapshotting > ------------------------------------------------------------------------ > > Key: FLINK-2098 > URL: https://issues.apache.org/jira/browse/FLINK-2098 > Project: Flink > Issue Type: Bug > Components: Streaming > Affects Versions: 0.9 > Reporter: Stephan Ewen > Assignee: Aljoscha Krettek > Priority: Blocker > Fix For: 0.9 > > > The stream source does not properly align the emission of checkpoint barriers > with the drawing of snapshots. -- This message was sent by Atlassian JIRA (v6.3.4#6332)