// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/CompositeTransactionManager.java
// public interface CompositeTransactionManager {
// 
// 	void open(String txId);
// 	void enlist(String txId, EntityCommand<?> ec);
// 	List<EntityCommand<?>> fetch(String txId);
// 	void close(String txId);
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/EntityCommand.java
// public class EntityCommand<T> implements Serializable{
// 	private static final long serialVersionUID = -3927305961526601453L;
// 
// 	public enum Action {
// 		INSERT, UPDATE, DELETE, QUERY
// 	}
// 
// 	private T entity;
// 	private Action action;
// 	private String transactionId;
// 	private long timestamp;
// 	
// 	public T getEntity() {
// 		return entity;
// 	}
// 
// 	public void setEntity(T entity) {
// 		this.entity = entity;
// 	}
// 
// 	public Action getAction() {
// 		return action;
// 	}
// 
// 	public void setAction(Action action) {
// 		this.action = action;
// 	}
// 
// 	public String getTransactionId() {
// 		return transactionId;
// 	}
// 
// 	public void setTransactionId(String transactionId) {
// 		this.transactionId = transactionId;
// 	}
// 
// 	public long getTimestamp() {
// 		return timestamp;
// 	}
// 
// 	public void setTimestamp(long timestamp) {
// 		this.timestamp = timestamp;
// 	}
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/Serializer.java
// public interface Serializer<T> {
// 
// 	byte[] write(T object) throws SerializationFailedException;
// 	String writeToString(T object) throws SerializationFailedException;
// 	
// 	T read(byte[] bytes) throws SerializationFailedException;
// 	T readFromString(String chars) throws SerializationFailedException;
// }


import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.EntityCommand;
import net.jotorren.microservices.tx.Serializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.kafka.core.KafkaTemplate;
package net.jotorren.microservices.tx.impl;




public class CompositeTransactionManagerKafkaImpl implements CompositeTransactionManager {

	private static final Logger LOG = LoggerFactory.getLogger(CompositeTransactionManager.class);

	@Value("${zookeeper.partitions:0}")
	private int zooPartitions;
	
	@Value("${zookeeper.replication:0}")
	private int zooReplication;
	
	@Value("${kafka.consumer.poll.timeout:0}")
	private long kafkaConsumerPollTimeout;
	
	
	@Autowired(required = false)
	private ZkUtils zkUtils;
	
	@Autowired(required = false)
	private KafkaTemplate<Integer, String> kafkaTemplate;

	@Autowired(required = false)
	@Qualifier("kafkaConsumerConfiguration")
	private Map<String, Object> configuration;
	
	@Autowired(required = false)
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/CompositeTransactionManager.java
// public interface CompositeTransactionManager {
// 
// 	void open(String txId);
// 	void enlist(String txId, EntityCommand<?> ec);
// 	List<EntityCommand<?>> fetch(String txId);
// 	void close(String txId);
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/EntityCommand.java
// public class EntityCommand<T> implements Serializable{
// 	private static final long serialVersionUID = -3927305961526601453L;
// 
// 	public enum Action {
// 		INSERT, UPDATE, DELETE, QUERY
// 	}
// 
// 	private T entity;
// 	private Action action;
// 	private String transactionId;
// 	private long timestamp;
// 	
// 	public T getEntity() {
// 		return entity;
// 	}
// 
// 	public void setEntity(T entity) {
// 		this.entity = entity;
// 	}
// 
// 	public Action getAction() {
// 		return action;
// 	}
// 
// 	public void setAction(Action action) {
// 		this.action = action;
// 	}
// 
// 	public String getTransactionId() {
// 		return transactionId;
// 	}
// 
// 	public void setTransactionId(String transactionId) {
// 		this.transactionId = transactionId;
// 	}
// 
// 	public long getTimestamp() {
// 		return timestamp;
// 	}
// 
// 	public void setTimestamp(long timestamp) {
// 		this.timestamp = timestamp;
// 	}
// }
// 
// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/Serializer.java
// public interface Serializer<T> {
// 
// 	byte[] write(T object) throws SerializationFailedException;
// 	String writeToString(T object) throws SerializationFailedException;
// 	
// 	T read(byte[] bytes) throws SerializationFailedException;
// 	T readFromString(String chars) throws SerializationFailedException;
// }


// Path: rahub-cross-cutting-concerns/src/main/java/net/jotorren/microservices/tx/impl/CompositeTransactionManagerKafkaImpl.java
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import kafka.admin.AdminUtils;
import kafka.admin.RackAwareMode;
import kafka.utils.ZkUtils;
import net.jotorren.microservices.tx.CompositeTransactionManager;
import net.jotorren.microservices.tx.EntityCommand;
import net.jotorren.microservices.tx.Serializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.kafka.core.KafkaTemplate;

package net.jotorren.microservices.tx.impl;




public class CompositeTransactionManagerKafkaImpl implements CompositeTransactionManager {

	private static final Logger LOG = LoggerFactory.getLogger(CompositeTransactionManager.class);

	@Value("${zookeeper.partitions:0}")
	private int zooPartitions;
	
	@Value("${zookeeper.replication:0}")
	private int zooReplication;
	
	@Value("${kafka.consumer.poll.timeout:0}")
	private long kafkaConsumerPollTimeout;
	
	
	@Autowired(required = false)
	private ZkUtils zkUtils;
	
	@Autowired(required = false)
	private KafkaTemplate<Integer, String> kafkaTemplate;

	@Autowired(required = false)
	@Qualifier("kafkaConsumerConfiguration")
	private Map<String, Object> configuration;
	
	@Autowired(required = false)
	private Serializer<EntityCommand<?>> serializer;