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!