/*
 * Copyright 2016 Red Hat Inc.
 *
 * 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.
 */

package enmasse.kafka.bridge.tracker;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

/**
 * Simple implementation of offset tracker.
 * It tracks only the offset for the last settled message. If receiver
 * settles "out of order", previous unsettled message won't be re-delivered
 * and MAY be lost so AT_LEAST_ONCE QoS is NOT guaranteed
 */
public class SimpleOffsetTracker<K, V> implements OffsetTracker<K, V> {

	// Apache Kafka topic to track
	private String topic;	
	// map with each partition and related tracked offset
	private Map<Integer, Long> offsets;
	// map with changed status of offsets
	private Map<Integer, Boolean> offsetsFlag;
	
	/**
	 * Contructor
	 *
	 * @param topic	topic to track offset
	 */
	public SimpleOffsetTracker(String topic) {
		
		this.topic = topic;
		this.offsets = new HashMap<>();
		this.offsetsFlag = new HashMap<>();
	}
	
	@Override
	public synchronized void track(int partition, long offset, ConsumerRecord<K, V> record) {
		// nothing
	}
	
	@Override
	public synchronized void delivered(int partition, long offset) {
		
		if (this.offsets.containsKey(partition)) {
			
			// map already contains partition but to handle "out of order" delivery
			// we have to check that the current delivered offset is greater than
			// the last committed offset
			if (offset > this.offsets.get(partition)) {
				this.offsets.put(partition, offset);
				this.offsetsFlag.put(partition, true);
			}
			
		} else {
			
			// new partition
			this.offsets.put(partition, offset);
			this.offsetsFlag.put(partition, true);
		}
	}

	@Override
	public synchronized Map<TopicPartition, OffsetAndMetadata> getOffsets() {
		
		Map<TopicPartition, OffsetAndMetadata> changedOffsets = new HashMap<>();
		
		for (Entry<Integer, Long> entry : this.offsets.entrySet()) {
			
			// check if partition offset is changed and it needs to be committed
			if (this.offsetsFlag.get(entry.getKey())) {
						
				changedOffsets.put(new TopicPartition(this.topic, entry.getKey()), 
						new OffsetAndMetadata(entry.getValue()));
			}
		}
		
		return changedOffsets;
	}
	
	@Override
	public synchronized void commit(Map<TopicPartition, OffsetAndMetadata> offsets) {
		
		for (Entry<TopicPartition, OffsetAndMetadata> offset : offsets.entrySet()) {
			
			// be sure we are tracking the current partition and related offset
			if (this.offsets.containsKey(offset.getKey().partition())) {
			
				// if offset tracked isn't changed during Kafka committing operation 
				// (it means no other messages were acknowledged)
				if (this.offsets.get(offset.getKey().partition()) == offset.getValue().offset()) {
					// we can mark this offset as committed (not changed)
					this.offsetsFlag.put(offset.getKey().partition(), false);
				}
			}
		}
	}

	@Override
	public synchronized void clear() {
		
		this.offsets.clear();
		this.offsetsFlag.clear();
	}
}

@@ -1201,16 +1201,220 @@
 , V%3E %7B%0A%0A
+%09private static class PartitionState %7B%0A%09%09public long offset;%0A%09%09public boolean changed;%0A%09%09public PartitionState(long offset, boolean changed) %7B%0A%09%09%09this.offset = offset;%0A%09%09%09this.changed = changed;%0A%09%09%7D%0A%09%7D%0A%09%0A
 %09// Apac
@@ -1542,91 +1542,22 @@
 er, 
-Long%3E offsets;%0A%09// map with changed status of offsets%0A%09private Map%3CInteger, Boolean
+PartitionState
 %3E of
@@ -1561,20 +1561,16 @@
  offsets
-Flag
 ;%0A%09%0A%09/**
@@ -1673,27 +1673,24 @@
 ng topic) %7B%0A
-%09%09%0A
 %09%09this.topic
@@ -1737,46 +1737,8 @@
 ();%0A
-%09%09this.offsetsFlag = new HashMap%3C%3E();%0A
 %09%7D%0A%09
@@ -2174,20 +2174,30 @@
 %0A%09%09%09
-if (offset %3E
+PartitionState state =
  thi
@@ -2224,94 +2224,93 @@
 ion)
-) %7B
+;
 %0A%09%09%09
-%09this.offsets.put(partition, 
+if (offset %3E state.
 offset)
-;
+ %7B
 %0A%09%09%09%09
-this
+state
 .offset
-sFlag.put(partition,
+ = offset;%0A%09%09%09%09state.changed =
  true
-)
 ;%0A%09%09
@@ -2386,57 +2386,41 @@
 on, 
-offset);%0A%09%09%09this.offsetsFlag.put(partition
+new PartitionState(offset
 , true)
+)
 ;%0A%09%09
@@ -2618,12 +2618,22 @@
 er, 
-Long
+PartitionState
 %3E en
@@ -2752,54 +2752,35 @@
 if (
-this.offsetsFlag.get(entry.getKey())) %7B%0A%09%09%09%09%09%09
+entry.getValue().changed) %7B
 %0A%09%09%09
@@ -2892,16 +2892,23 @@
 tValue()
+.offset
 ));%0A%09%09%09%7D
@@ -3396,20 +3396,39 @@
 ed)%0A%09%09%09%09
-if (
+PartitionState state = 
 this.off
@@ -3464,16 +3464,38 @@
 ition())
+;%0A%09%09%09%09if (state.offset
  == offs
@@ -3588,64 +3588,29 @@
 %09%09%09%09
-this.offsetsFlag.put(offset.getKey().partition(),
+state.changed =
  false
-)
 ;%0A%09%09
@@ -3704,36 +3704,8 @@
 ();%0A
-%09%09this.offsetsFlag.clear();%0A
 %09%7D%0A%7D

Refactor SimpleOffsetTracker to use a single Map