Aleph 70 kafka bugs

aleph2_core_distributed_services_library/src/com/ikanow/aleph2/distributed_services/utils/KafkaUtils.java

@@ -15,6 +15,7 @@
 ******************************************************************************/
 package com.ikanow.aleph2.distributed_services.utils;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -63,7 +64,9 @@ public class KafkaUtils {
 	private static Properties kafka_properties = new Properties();
 	private final static Logger logger = LogManager.getLogger();
 	protected final static Map<String, Boolean> my_topics = new ConcurrentHashMap<String, Boolean>(); // (Things to which I am publishing)
-	protected final static Cache<String, Boolean> known_topics = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).build();
+	protected final static Cache<String, Boolean> known_topics = CacheBuilder.newBuilder().expireAfterWrite(5, TimeUnit.MINUTES).build();	
+	protected static int producer_pool_index = -1;
+	protected static List<Producer<String,String>> producer_pool = null;
 	//TODO (ALEPH-12): make my_topics a cached map also
 
 	/** Creates a new ZK client from the properties
@@ -81,16 +84,23 @@ public synchronized static ZkClient getNewZkClient() {
 	 * 
 	 * @return
 	 */
-	public synchronized static Producer<String, String> getKafkaProducer() {		
-		if ( producer == null ) {
+	public synchronized static Producer<String, String> getKafkaProducer() {	
+		final int num_producers = 25; //TODO make this configurable, probably per topic rather than globally?
+		if ( producer_pool == null ) {
+			producer_pool = new ArrayList<Producer<String,String>>(num_producers);
 			ProducerConfig config = new ProducerConfig(kafka_properties);
-			producer = new Producer<String, String>(config);
+			for ( int i = 0; i < num_producers; i++ ) {
+				producer_pool.add(new Producer<String, String>(config));
+			}
 		}
-        return producer;
+		producer_pool_index = (producer_pool_index+1)%num_producers;
+        return producer_pool.get(producer_pool_index);
 	}
 
 	/**
 	 * Creates a consumer for a single topic with the currently configured Kafka instance.
+	 * WARNING: When a consumer is created, it starts its reading at now, so if you
+	 * previously produced on a topic, this consumer won't be able to see it.
 	 * 
 	 * This consumer should be closed once you are done reading.
 	 * 
@@ -107,6 +117,7 @@ public static ConsumerConnector getKafkaConsumer(String topic, Optional<String>
 						final Properties np = new Properties();
 						kafka_properties.forEach((key, val) -> np.put(key, val));
 						np.put("group.id", name);
+						//np.put("auto.offset.reset", "largest");
 						return np;
 					})
 				.orElse(kafka_properties)
@@ -170,10 +181,8 @@ public static boolean doesTopicExist(final String topic, final ZkClient zk_clien
 	public static void setProperties(Config parseMap) {
 		kafka_properties = new Properties();
 		final Map<String, Object> config_map_kafka = ImmutableMap.<String, Object>builder()
-				.put("group.id", "aleph2_unknown")
-				.put("serializer.class", "kafka.serializer.StringEncoder")
-				.put("request.required.acks", "1")
-				.put("consumer.timeout.ms", "3000")
+				.put("group.id", "aleph2_unknown")				
+//				.put("consumer.timeout.ms", "3000") //this determines how long a consumer.hasNext() will wait before crashing out (see WrappedConsumerIterator)
 		        .put("auto.commit.interval.ms", "1000")
 		        // Not sure which of these 2 sets is correct, so will list them both!
 		        // these are listed here: https://kafka.apache.org/08/configuration.html
@@ -185,18 +194,29 @@ public static void setProperties(Config parseMap) {
 				.put("zk.sessiontimeout.ms", "6000")
 				.put("zk.synctime.ms", "2000")
 				.put("delete.topic.enable", "true")
+
+				//producer specific config		
+				.put("serializer.class", "kafka.serializer.StringEncoder")
+				.put("request.required.acks", "1")
+				.put("producer.type", "async")
+				.put("compression.codec", "2")
+				.put("batch.num.messages", "800")
 				.build();	
 
 		final Config fullConfig = parseMap.withFallback(ConfigFactory.parseMap(config_map_kafka));
 		fullConfig.entrySet().stream().forEach(e -> kafka_properties.put(e.getKey(), e.getValue().unwrapped()));
 
 		//PRODUCER PROPERTIES		
-		String broker = fullConfig.getString("metadata.broker.list");
-		logger.debug("BROKER: " + broker);
+		if ( fullConfig.hasPath("metadata.broker.list") ) {
+			String broker = fullConfig.getString("metadata.broker.list");
+			logger.debug("BROKER: " + broker);
+		}
 
-        //CONSUMER PROPERTIES		
-		String zk = fullConfig.getString("zookeeper.connect");
-		logger.debug("ZOOKEEPER: " + zk);
+        //CONSUMER PROPERTIES
+		if ( fullConfig.hasPath("zookeeper.connect") ) {
+			String zk = fullConfig.getString("zookeeper.connect");
+			logger.debug("ZOOKEEPER: " + zk);
+		}
 
         //reset producer so a new one will be created
         if ( producer != null )
@@ -268,7 +288,8 @@ public synchronized static void createTopic(String topic, Optional<Map<String, O
 				logger.debug("LEADER WAS ELECTED: " + leader_elected);
 
 				//create a consumer to fix offsets (this is a hack, idk why it doesn't work until we create a consumer)
-				WrappedConsumerIterator iter = new WrappedConsumerIterator(getKafkaConsumer(topic, Optional.empty()), topic);
+				//timeout is set to 1ms to immediately crash out, no need to waste time, it'll just block forever
+				WrappedConsumerIterator iter = new WrappedConsumerIterator(getKafkaConsumer(topic, Optional.empty()), topic, 1);
 				iter.hasNext();
 
 				//debug info

aleph2_core_distributed_services_library/src/com/ikanow/aleph2/distributed_services/utils/WrappedConsumerIterator.java

@@ -20,6 +20,12 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -41,6 +47,7 @@ public class WrappedConsumerIterator implements Closeable, Iterator<String> {
 	final protected String topic;	
 	final protected Iterator<MessageAndMetadata<byte[], byte[]>> iterator;
 	final private static Logger logger = LogManager.getLogger();
+	final protected long force_timeout_ms;
 
 	/**
 	 * Takes a consumer and the topic name, retrieves the stream of results and
@@ -50,14 +57,19 @@ public class WrappedConsumerIterator implements Closeable, Iterator<String> {
 	 * @param topic
 	 */
 	public WrappedConsumerIterator(ConsumerConnector consumer, String topic) {
+        this(consumer, topic, 0);
+	}
+
+	public WrappedConsumerIterator(ConsumerConnector consumer, String topic, long force_timeout_ms) {
 		this.consumer = consumer;
 		this.topic = topic;
 		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
         topicCountMap.put(topic, 1);
         final Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
         final List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
         final KafkaStream<byte[], byte[]> stream = streams.get(0);
-        this.iterator = stream.iterator();       
+        this.iterator = stream.iterator();    
+        this.force_timeout_ms = force_timeout_ms;
 	}
 
 	/**
@@ -83,13 +95,41 @@ public String next() {
 	 * that timeout and return false, otherwise it will block forever until a new item is found,
 	 * it never returns false from the internal iterator, we do on an exception (timeout)
 	 * 
+	 * If force_timeout_ms is set, will only wait a max of it for hasNext to return, if set to 0 or less, will
+	 * just leave it up to kafka config for when to kick out of hasNext (see consumer.timeout.ms)
+	 * 
 	 */
 	@Override
 	public boolean hasNext() {
+		final ExecutorService executor = Executors.newSingleThreadExecutor();
+		Future<Boolean> future = executor.submit(new Callable<Boolean>() {
+			@Override
+			public Boolean call() throws Exception {
+				try {
+					return iterator.hasNext();
+				} catch (Exception e) {
+					logger.debug("Topic iterator exceptioned (typically because no item was found in timeout period), this is set in KafkaUtils via consumer.timeout.ms", e);
+					close();
+					return false;
+				}	
+			}
+		});
+		executor.shutdown();
+		if ( force_timeout_ms > 0 ) {
+			try {
+				executor.awaitTermination(force_timeout_ms, TimeUnit.MILLISECONDS);				
+			} catch (Exception ex) {
+				logger.debug("Topic iterator exceptioned (typically because no item was found in timeout period), this is set in KafkaUtils via consumer.timeout.ms", ex);
+				close();
+				return false;
+			} finally {
+				executor.shutdownNow();
+			}
+		}
 		try {
-			return iterator.hasNext();
-		} catch (Exception e) {
-			logger.debug("Topic iterator exceptioned (typically because no item was found in timeout period), this is set in KafkaUtils via consumer.timeout.ms");
+			return future.get();
+		} catch (InterruptedException | ExecutionException e) {
+			logger.debug("Topic iterator exceptioned (typically because no item was found in timeout period), this is set in KafkaUtils via consumer.timeout.ms", e);
 			close();
 			return false;
 		}		

aleph2_core_distributed_services_library/test/com/ikanow/aleph2/distributed_services/services/TestCoreDistributedServices.java

@@ -184,6 +184,7 @@ public void testKafka() throws Exception {
 		for ( int i = 0; i < num_to_test; i++ ) {
 			_core_distributed_services.produce(TOPIC_NAME, original_message);
 		}
+		Thread.sleep(5000); //wait a few seconds for producers to dump batch
 
 		//grab the consumer
 		Iterator<String> consumer = _core_distributed_services.consumeAs(TOPIC_NAME, Optional.empty());
@@ -224,6 +225,7 @@ public void testKafkaForStormSpout() throws Exception {
 		String original_message = jsonNode.toString();	
 		for ( int i = 0; i < num_to_test; i++ ) 
 			_core_distributed_services.produce(TOPIC_NAME, original_message);	
+		Thread.sleep(10000); //wait a few seconds for producers to dump batch
 
 		//grab the consumer
 		Iterator<String> consumer = _core_distributed_services.consumeAs(TOPIC_NAME, Optional.empty());

aleph2_core_distributed_services_library/test/com/ikanow/aleph2/distributed_services/utils/TestKafkaUtils.java

@@ -22,6 +22,7 @@
 import kafka.javaapi.consumer.ConsumerConnector;
 import kafka.javaapi.producer.Producer;
 import kafka.producer.KeyedMessage;
+import kafka.utils.ZkUtils;
 
 import org.I0Itec.zkclient.ZkClient;
 import org.junit.Before;
@@ -104,7 +105,7 @@ public void testCreateTopic() throws InterruptedException {
 		final String topic = "test_create";
 		final ZkClient zk_client = KafkaUtils.getNewZkClient();
 		KafkaUtils.createTopic(topic, Optional.empty(), zk_client);		
-		Thread.sleep(5000);
+//		Thread.sleep(5000);
 		assertTrue(KafkaUtils.doesTopicExist(topic, zk_client));
 	}
 
@@ -113,25 +114,25 @@ public void testCreateTopic() throws InterruptedException {
 	 * 
 	 * @throws InterruptedException
 	 */
+//	@Ignore
 	@Test
 	public void testProduceConsume() throws InterruptedException {
 		final String topic = "test_produce_consume";
 		final ZkClient zk_client = KafkaUtils.getNewZkClient();
 		KafkaUtils.createTopic(topic, Optional.empty(), zk_client);		
-		Thread.sleep(5000);
+//		Thread.sleep(5000);
 		assertTrue(KafkaUtils.doesTopicExist(topic, zk_client));
 
 		//write something into the topic
 		Producer<String, String> producer = KafkaUtils.getKafkaProducer();
 		long num_messages_to_produce = 5;
 		for (long i = 0; i < num_messages_to_produce; i++)
-			producer.send(new KeyedMessage<String, String>(topic, "test"));		
-
-		Thread.sleep(5000);
+			producer.send(new KeyedMessage<String, String>(topic, "test"));				
+		Thread.sleep(5000); //sleep to wait for records getting moved
 
 		//see if we can read that items
 		ConsumerConnector consumer2 = KafkaUtils.getKafkaConsumer(topic, Optional.empty());
-		WrappedConsumerIterator wrapped_consumer2 = new WrappedConsumerIterator(consumer2, topic);		
+		WrappedConsumerIterator wrapped_consumer2 = new WrappedConsumerIterator(consumer2, topic, 2000);		
 		long count = 0;
 		while ( wrapped_consumer2.hasNext() ) {
 			wrapped_consumer2.next();
@@ -150,24 +151,25 @@ public void testProduceConsume() throws InterruptedException {
 	 * 
 	 * @throws InterruptedException
 	 */
-	@Ignore
+	@Ignore //Currently ignored because local delete fails, see log output for error messages
 	@Test
-	public void testDeleteTopic() throws InterruptedException {
-		final String topic = "test_delete_topic11";
+	public void testDeleteTopic() throws InterruptedException {		
+		final String topic = "test_delete_topic";
 		final ZkClient zk_client = KafkaUtils.getNewZkClient();
 
 		//Create a topic to delete later
 		KafkaUtils.createTopic(topic, Optional.empty(), zk_client);
 
-		Thread.sleep(5000);
+//		Thread.sleep(5000);
 
 		//write something into the topic
 		Producer<String, String> producer = KafkaUtils.getKafkaProducer();
 		long num_messages_to_produce = 3;
-		for (long i = 0; i < num_messages_to_produce; i++)
-			producer.send(new KeyedMessage<String, String>(topic, "test"));		
-
-		Thread.sleep(5000);
+		for (long i = 0; i < num_messages_to_produce; i++) {
+			System.out.println("producing message: " + i);
+			producer.send(new KeyedMessage<String, String>(topic, "test"));
+		}
+		Thread.sleep(5000); //sleep to wait for records getting moved
 
 		//delete the topic
 		assertTrue(KafkaUtils.doesTopicExist(topic, zk_client));
@@ -183,7 +185,7 @@ public void testDeleteTopic() throws InterruptedException {
 		System.out.println("STARTING TO GET CONSUMER");
 		//see if we can read that iem
 		ConsumerConnector consumer1 = KafkaUtils.getKafkaConsumer(topic, Optional.empty());
-		WrappedConsumerIterator wrapped_consumer1 = new WrappedConsumerIterator(consumer1, topic);
+		WrappedConsumerIterator wrapped_consumer1 = new WrappedConsumerIterator(consumer1, topic, 2000);
 		System.out.println("LOOPING OVER MESSAGES");
 		while ( wrapped_consumer1.hasNext() ) {
 			System.out.println("NEXT: " + wrapped_consumer1.next());
@@ -212,4 +214,88 @@ public void testDeleteNonExistantTopic() {
 		}
 		assertFalse(KafkaUtils.my_topics.containsKey(topic));
 	}
+
+	/**
+	 * Tests creating a named consumer, then closing it and cleaning it up.
+	 * 1. Create topic
+	 * 2. Create consumer
+	 * 3. Produce some data
+	 * 4. Consume said data with previous consumer
+	 * 5. Close consumer, assert it doesnt exist
+	 * 6. Open consumer with same name
+	 * 7. Produce some data
+	 * 8. Consume said data
+	 * 9. Close consumer
+	 * @throws InterruptedException 
+	 */
+	@Ignore //This test currently fails when run with the group for some reason?
+	//it also isn't really doing anything currently because kafka doesn't fully cleanup consumers
+	//in ZK currently.
+	@Test
+	public void testConsumerCleanup() throws InterruptedException {
+		final String topic = "test_consumer_cleanup";
+		final String group_id = "test_consumer";
+		final ZkClient zk_client = KafkaUtils.getNewZkClient();
+
+		System.out.println("CREATING TOPIC");
+		KafkaUtils.createTopic(topic, Optional.empty(), zk_client);		
+		//Thread.sleep(5000);
+		assertTrue(KafkaUtils.doesTopicExist(topic, zk_client));
+
+		System.out.println("CREATING CONSUMER");
+		//create a named consumer before we start producing
+		ConsumerConnector consumer = KafkaUtils.getKafkaConsumer(topic, Optional.of(group_id));
+		@SuppressWarnings("resource")
+		WrappedConsumerIterator wrapped_consumer = new WrappedConsumerIterator(consumer, topic, 2000);
+
+		System.out.println("PRODUCE SOME DATA");
+		//write something into the topic
+		Producer<String, String> producer = KafkaUtils.getKafkaProducer();
+		long num_messages_to_produce = 5;
+		for (long i = 0; i < num_messages_to_produce; i++) {
+			System.out.println("produce message: " + i);
+			producer.send(new KeyedMessage<String, String>(topic, "test_pt1_" + i));
+		}
+		Thread.sleep(5000); //sleep to wait for records getting moved
+
+		System.out.println("CONSUMING DATA");
+		//see if we can read that items
+		long count = 0;
+		while ( wrapped_consumer.hasNext() ) {
+			wrapped_consumer.next();
+			count++;
+		}
+		assertEquals(count, num_messages_to_produce);
+
+		System.out.println("DELETING CONSUMER");
+		//assert consumer exists
+		assertTrue(ZkUtils.pathExists(zk_client, ZkUtils.ConsumersPath() + "/" + group_id));
+		//close consumer
+		wrapped_consumer.close();
+		//assert consumer no longer exists
+		//NOTE: current consumer does not delete this entry out, you have to manually handle it
+		//we could delete it via ZKUtils.deletePathRecursively but waiting until 0.8.2 to see how that handles
+		//TODO when we want to fully kill consumers we can put this line back in
+		//assertFalse(ZkUtils.pathExists(zk_client, ZkUtils.ConsumersPath() + "/" + group_id));
+
+		System.out.println("CREATING CONSUMER AGAIN, REUSING NAME");
+		consumer = KafkaUtils.getKafkaConsumer(topic, Optional.of(group_id));
+		wrapped_consumer = new WrappedConsumerIterator(consumer, topic, 2000);	
+
+		System.out.println("PRODUCE SOME DATA");
+		//assert we can reuse the same consumer
+		//write something into the topic, again
+		for (long i = 0; i < num_messages_to_produce; i++)
+			producer.send(new KeyedMessage<String, String>(topic, "test_pt2"));				
+		Thread.sleep(5000); //sleep to wait for records getting moved
+
+		System.out.println("CONSUME DATA");
+		//see if we can read that items			
+		count = 0;
+		while ( wrapped_consumer.hasNext() ) {
+			wrapped_consumer.next();
+			count++;
+		}
+		assertEquals(count, num_messages_to_produce);
+	}
 }