Twitter API is not working for Kafka. it's refusing the connection

[ashvani1]

import java.util.List; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit;

import org.slf4j.Logger; import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists; import com.twitter.hbc.ClientBuilder; import com.twitter.hbc.core.Client; import com.twitter.hbc.core.Constants; import com.twitter.hbc.core.Hosts; import com.twitter.hbc.core.HttpHosts; import com.twitter.hbc.core.endpoint.StatusesFilterEndpoint; import com.twitter.hbc.core.processor.StringDelimitedProcessor; import com.twitter.hbc.httpclient.auth.Authentication; import com.twitter.hbc.httpclient.auth.OAuth1;

public class TwitterProducer { Logger logger = LoggerFactory.getLogger(TwitterProducer.class.getName()); String consumerKey = "V1pedBFKWPM2zXqzoFzUOPNrL"; String consumerSecret = "2RsTvxFtZVZFe2FFU2KGqLglhPu8cia2JvTXPCYRfUswP36Td6"; String token = "555591025-555591025-MrPsbVxBNiNrAUBnkzHKYy8AnJahRtzHsjSZwzqz"; String secret = "7F4Y9PYzAdpg0l90BeaLOBnr5Jn2sIMxNz1HK4OMYAOaK";

public TwitterProducer() {}

public static void main(String[] args)
{
    new TwitterProducer().run();
}

public void run()
{
            logger.info("Setup");

    /** Set up your blocking queues: Be sure to size these properly based on expected TPS of your stream */
    BlockingQueue<String> msgQueue = new LinkedBlockingQueue<String>(10);
    //create a twitter client
    Client client = createTwitterClient(msgQueue);
    // Attempts to establish a connection.
    client.connect();

    //create a kafka producer
    //loop to send tweets to kafka
    // on a different thread, or multiple different threads....
    while (!client.isDone()) {
        String msg = null;
      try {
           msg = msgQueue.poll(5, TimeUnit.SECONDS);
      }
      catch(InterruptedException e)
      {
          e.printStackTrace();
          client.stop(); 
      }

      if(msg != null) {
          logger.info(msg);
      }
    }
    logger.info("End of Application");
}

public Client createTwitterClient(BlockingQueue<String> msgQueue)
{

    /** Declare the host you want to connect to, the endpoint, and authentication (basic auth or oauth) */
    Hosts hosebirdHosts = new HttpHosts(Constants.STREAM_HOST);
    StatusesFilterEndpoint hosebirdEndpoint = new StatusesFilterEndpoint();
    List<String> terms = Lists.newArrayList("ashvani");

    hosebirdEndpoint.trackTerms(terms);

    // These secrets should be read from a config file
    Authentication hosebirdAuth = new OAuth1(consumerKey, consumerSecret, token, secret);

            ClientBuilder builder = new ClientBuilder()
                      .name("Hosebird-Client-01")                              // optional: mainly for the logs
                      .hosts(hosebirdHosts)
                      .authentication(hosebirdAuth)
                      .endpoint(hosebirdEndpoint)
                      .processor(new StringDelimitedProcessor(msgQueue));

                    Client hosebirdClient = builder.build();
                    return hosebirdClient;

}

}

[ubarpsi]

Hope, you have found the answer to this already. If not, please try the below steps for authentication; it will work: // sample code // generate the OAuth Access token first and then set it with twitter handle twitter = new TwitterFactory().getInstance(); twitter.setOAuthConsumer(CONSUMER_KEY, CONSUMER_SECRET_KEY); AccessToken oauthAccessToken = new AccessToken(getSavedAccessToken(), getSavedAccessTokenSecret()); twitter.setOAuthAccessToken(oauthAccessToken);

I assume, you have saved the access Token and access token secret from our twitter App - permissions page already. Use them to supply getSavedAccessToken() getSavedAccessTokenSecret()

Hope this helps.

[EDKarlsson]

This should be resolved, @ubarpsi solution worked!