<repo_name>patrickmcdonogh/dhlconsumer<file_sep>/java/src/dhl/demo/MessageHandler.java
package dhl.demo;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.RangeAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.apama.epl.plugin.Correlator;
import com.apama.util.Logger;

public class MessageHandler implements KafkaMessageHandler {
	final static Logger logger = Logger.getLogger();

	KafkaConsumer<String, String> consumer;

	public MessageHandler() {
	}

	@Override
	public int pollAndHandleMessages() {
		ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(300)); 
		int recordCount = records.count();

		if (recordCount > 0) {
			for (ConsumerRecord<String, String> record : records) {
				try {
					logger.info("Managing message = "+record.value());
					Correlator.sendTo(record.value(), "");
				}
				catch(Exception ex) {
					logger.error("Issue sending data to correlator", ex);
				}
			}
		}
		return recordCount;
	}

	@Override
	public void setupKafkaConsumer(String bootstrapServers, String topic, String consumerGroup, boolean autoAck, ConsumerRebalanceListener rebalanceManager) {
		Properties properties = new Properties();
		properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
		properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
		String valueDeserializer = StringDeserializer.class.getName();
		properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
		properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
		properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, String.valueOf(autoAck));
		properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
		properties.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RangeAssignor.class.getName());
		properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
		properties.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1280000");

		//switch classloader so that kafka can find the serializers
		ClassLoader original = Thread.currentThread().getContextClassLoader();
		Thread.currentThread().setContextClassLoader(null);

		consumer = new KafkaConsumer<String, String>(properties);

		Thread.currentThread().setContextClassLoader(original);
		consumer.subscribe(Arrays.asList(topic), rebalanceManager);
	}

	@Override
	public void terminatePoll() {
		consumer.unsubscribe();
		consumer.close();
	}

}
<file_sep>/java/src/dhl/demo/TopicProducer.java
package dhl.demo;

import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.serialization.StringSerializer;

import com.apama.util.Logger;

public class TopicProducer<K,V>  {
	
	protected static Logger logger = Logger.getLogger();
	
	protected String bootstrapServers;
	protected String topic;
	
	protected KafkaProducer<K, V> producer;
	private BlockingQueue<ProducerRecord<K, V>> messageQueue;
	
	private int messagesPublished;
	
	private final AtomicBoolean running = new AtomicBoolean(true);
	
	public TopicProducer(String bootstrapServers, String topic) {
		super();
		this.bootstrapServers = bootstrapServers;
		this.topic = topic;
		createPublisher();
	}

	public String getTopic() {
		return topic;
	}
	
	private void createPublisher() {
        //switch classloader so that kafka can find the serializers
        ClassLoader original = Thread.currentThread().getContextClassLoader();
        Thread.currentThread().setContextClassLoader(null);

		producer = createProducer();
		logger.info("Created producer for " + topic);

        Thread.currentThread().setContextClassLoader(original);
        
        messageQueue = new ArrayBlockingQueue<>(200 * 1000);
        
        new Thread(new Runnable() {
			
			@Override
			public void run() {
				while(running.get()) {
					try {
						sendRecord((ProducerRecord<K, V>)messageQueue.take());
				        messagesPublished++;
					}
					catch(Exception ex) {
						logger.error("Error sending data to Kafka topic = "+topic, ex);
					}
				}
				
				logger.info("Closing producer on topic = "+topic);
				producer.close();
				
			}
		}).start();
	}

	protected KafkaProducer<K, V> createProducer() {
		Properties properties = new Properties();
		properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "60000");
        properties.setProperty(ProducerConfig.ACKS_CONFIG, "all");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG, "10");
        properties.setProperty(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "2097152");
        properties.setProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "300000");

		return new KafkaProducer<K, V>(properties);
	}

	protected void sendRecord(ProducerRecord<K, V> record) {		
		ProducerRecord<K, V> theRecord = (ProducerRecord<K, V>)record;
			
		producer.send(theRecord, new Callback() {
			@Override
			public void onCompletion(RecordMetadata arg0, Exception ex) {
				if (ex != null) {
					logger.error("Error sending message to Kafka topic = "+topic, ex);
					if (ex instanceof RecordTooLargeException) {
						if (theRecord.value() instanceof String) {
							logger.error("Too Large Message:" + ((String)theRecord.value()).substring(0,500));
						}
						
					}
					else {
						logger.error("Message:" + record.value());
					}
				}
			}
		});
	}
	
	public void stop() {
		running.set(false);
	}
	
	public boolean publish(ProducerRecord<K, V> record, boolean block) {
		boolean output = true;
		try {
			if ( block ) {
				messageQueue.put(record);
			}
			else
			{				
				output = messageQueue.offer(record);
		        if ( !output ) {
	        	logger.error("Could not add message to internal queue for Kafka topic = "+topic+". Internal message queue size = "+messageQueue.size());
	        }
		}
		}
		catch (Exception ex) {
			logger.error("Error sending message to topic = "+topic, ex);
			output = false;
		}
		return output;
	}
	
}
<file_sep>/Dockerfile
# Copyright (c) 2017-2018 Software AG, Darmstadt, Germany and/or its licensors
#
# Licensed 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.

# Create an image from the base Apama image, copying the project in it ready to deploy
ARG APAMA_IMAGE=store/softwareag/apama-correlator:10.5
FROM ${APAMA_IMAGE}

COPY --chown=sagadmin:sagadmin ApamaServerLicense.xml ${APAMA_WORK}/DHLDemo/
COPY --chown=sagadmin:sagadmin init.yaml ${APAMA_WORK}/DHLDemo/
COPY --chown=sagadmin:sagadmin eventdefinitions ${APAMA_WORK}/DHLDemo/eventdefinitions
COPY --chown=sagadmin:sagadmin monitors ${APAMA_WORK}/DHLDemo/monitors
COPY --chown=sagadmin:sagadmin plugin ${APAMA_WORK}/DHLDemo/plugin
COPY --chown=sagadmin:sagadmin lib ${APAMA_WORK}/lib

WORKDIR ${APAMA_WORK}

CMD ["correlator", "--config", "DHLDemo"]
<file_sep>/java/src/dhl/demo/KafkaMessageHandler.java
package dhl.demo;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;

public interface KafkaMessageHandler {
	int pollAndHandleMessages();
	void setupKafkaConsumer(String bootstrapServers, String topic, String consumerGroup, boolean autoAck, ConsumerRebalanceListener rebalanceManager);
	void terminatePoll();
}
