package org.apache.kafka.streams.state;

public class MockKeyValueStore<K, V>
        extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, K, V>
        implements KeyValueStore<K, V>  {
    // keep a global counter of flushes and a local reference to which store had which
    // flush, so we can reason about the order in which stores get flushed.
    private static final AtomicInteger GLOBAL_FLUSH_COUNTER = new AtomicInteger(0);
    private final AtomicInteger instanceLastFlushCount = new AtomicInteger(-1);


    public boolean initialized = false;
    public boolean flushed = false;
    public boolean closed = true

package org.apache.kafka.streams.internals;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateRestoreCallback;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

public class MockKeyValueStore<K, V> implements KeyValueStore<K, V>  {
    // keep a global counter of flushes and a local reference to which store had which
    // flush, so we can reason about the order in which stores get flushed.
    private static final AtomicInteger GLOBAL_FLUSH_COUNTER = new AtomicInteger(0);
    private final AtomicInteger instanceLastFlushCount = new AtomicInteger(-1);
    private final String name;
    private final boolean persistent;

    public boolean initialized = false;
    public boolean flushed = false;
    public boolean closed = true;

    public final List<KeyValue<K, V>> capturedPutCalls = new LinkedList<>();
    public final List<KeyValue<K, V>> capturedGetCalls = new LinkedList<>();
    public final List<KeyValue<K, V>> capturedDeleteCalls = new LinkedList<>();

    public final KeyValueStore<K, V> innerKeyValueStore;

    public MockKeyValueStore(final String name,;
    public boolean persistent;

    protected final Time time;


    final Serde<K> keySerde;
    final Serde<V> valueSerde;
    final KeyValueStore<KStateSerdes<K, V> innerKeyValueStore,serdes;

    public final List<KeyValue<K, V>> capturedPutCalls = new LinkedList<>();
    public final List<KeyValue<K, V>> capturedGetCalls = new LinkedList<>();
    public final  final boolean persistent) {
List<KeyValue<K, V>> capturedDeleteCalls = new LinkedList<>();


    public MockKeyValueStore(final KeyValueBytesStoreSupplier keyValueBytesStoreSupplier,
  this.name = name;
        this.innerKeyValueStore = innerKeyValueStore;
        this.persistent = persistent;

    }

  final Serde<K> @OverridekeySerde,
    public String name() {
        return name;
    }

    @Override
    public void init(final ProcessorContextSerde<V> contextvalueSerde,
                     final StateStore root) {
     final   context.register(root, stateRestoreCallback);
boolean persistent,
         initialized = true;
        closed = false;
    }

    @Override
final    public void flush(Time time) {
        instanceLastFlushCount.set(GLOBAL_FLUSH_COUNTER.getAndIncrementsuper(keyValueBytesStoreSupplier.get());
         innerKeyValueStore.flushthis.name = keyValueBytesStoreSupplier.name();
        flushedthis.time = true;
time != null ? }

time : Time.SYSTEM;
  public int getLastFlushCount() {
   this.persistent = persistent;
      return instanceLastFlushCount.get()  this.keySerde = keySerde;
        this.valueSerde = valueSerde;
    }

    @Override
 @SuppressWarnings("unchecked")
   public void closeinitStoreSerde(final ProcessorContext context) {
        innerKeyValueStore.close();serdes = new StateSerdes<>(
        closed = true;
    }

    @OverrideProcessorStateManager.storeChangelogTopic(context.applicationId(), name()),
    public boolean persistent() {
        return persistent;
keySerde == null  }

    @Override? (Serde<K>) context.keySerde() : keySerde,
    public boolean isOpen() {
        return !closed;
    valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
    }

    public@Override
 final StateRestoreCallback stateRestoreCallback =public newString StateRestoreCallbackname() {

        @Overridereturn name;
    }

    @Override
    public void restoreinit(final byte[]ProcessorContext keycontext,
                     final StateStore root) {
     final byte[] value) {  context.register(root, stateRestoreCallback);
        initialized = true;
        }closed = false;
    };

    @Override
    public void put(final K key, final V valueflush() {
        capturedPutCallsinstanceLastFlushCount.add(new KeyValue<>(key, valueset(GLOBAL_FLUSH_COUNTER.getAndIncrement());
        innerKeyValueStorewrapped().putflush(key, value);
      }

  flushed = @Overridetrue;
    public}

 V putIfAbsent(final K key,public final V valueint getLastFlushCount() {
        final V originalValue = return instanceLastFlushCount.get(key);
    }

    @Override
   if (originalValuepublic ==void nullclose() {
            put(key, valuewrapped().close();
        closed = true;
  capturedPutCalls.add(new KeyValue<>(key, value));
 }

    @Override
     }public boolean persistent() {
        return originalValuepersistent;
    }

    @Override
    public Vboolean deleteisOpen(final K key) {
        V value = innerKeyValueStore.delete(key)return !closed;
     }

   capturedDeleteCalls.add(new KeyValue<>(key, value));
        return value;
 public final StateRestoreCallback stateRestoreCallback = new StateRestoreCallback() {
        }@Override

    @Override
    public void putAllrestore(final List<KeyValue<Kbyte[] key,
 V>> entries) {
        for (final KeyValue<K, V> entry : entries) {
          final  put(entry.key, entry.value);
            capturedPutCalls.add(entry);byte[] value) {
        }
    };

    @Override
    public Vvoid getput(final K key), {
        final V value = innerKeyValueStore.get(key);) {
        capturedGetCallscapturedPutCalls.add(new KeyValue<>(key, value));
        return value;wrapped().put(keyBytes(key), serdes.rawValue(value));
    }

    @Override
    public KeyValueIterator<K,V>V rangeputIfAbsent(final K fromkey, final KV tovalue) {
        return innerKeyValueStore.range(from, tofinal V originalValue = get(key);
    }

    @Override
if (originalValue == null) public KeyValueIterator<K,V>  all() {
{
           return innerKeyValueStore.all(put(key, value);
      }

      @Override
capturedPutCalls.add(new KeyValue<>(key, value));
     public long approximateNumEntries() {}
        return innerKeyValueStore.approximateNumEntries()originalValue;
    }
}