package

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.StateStorestate;

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>  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);
    private final String name;
    private final boolean persistent;

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

    public String finalname;
 List<KeyValue<K, V>> capturedPutCalls =public newboolean LinkedList<>()persistent;

    protected final Time time;


    final Serde<K> keySerde;
    final Serde<V> valueSerde;
    StateSerdes<K, V> serdes;

    public final List<KeyValue<K, V>> capturedGetCallscapturedPutCalls = new LinkedList<>();
    public final List<KeyValue<K, V>> capturedDeleteCallscapturedGetCalls = new LinkedList<>();

    public final KeyValueStore<KList<KeyValue<K, V> innerKeyValueStoreV>> capturedDeleteCalls = new LinkedList<>();


    public MockKeyValueStore(final StringKeyValueBytesStoreSupplier namekeyValueBytesStoreSupplier,
                             final KeyValueStore<K, V> innerKeyValueStoreSerde<K> keySerde,
                             final boolean persistent) {
Serde<V> valueSerde,
         this.name = name;
        this.innerKeyValueStore = innerKeyValueStore;
        this.persistentfinal =boolean persistent;
,
    }

    @Override
    public String name() {
        return name;
    }

 final Time time) @Override{
    public void init(final ProcessorContext context, super(keyValueBytesStoreSupplier.get());
        this.name = keyValueBytesStoreSupplier.name();
        this.time = time != finalnull ? StateStoretime root) {: Time.SYSTEM;
        context.register(root, stateRestoreCallback)this.persistent = persistent;
        initializedthis.keySerde = truekeySerde;
        closedthis.valueSerde = falsevalueSerde;
    }

    @Override@SuppressWarnings("unchecked")
    public void flushinitStoreSerde(final ProcessorContext context) {
        instanceLastFlushCount.set(GLOBAL_FLUSH_COUNTER.getAndIncrement());
serdes = new StateSerdes<>(
          innerKeyValueStore.flush();      ProcessorStateManager.storeChangelogTopic(context.applicationId(), name()),
        flushed = true;
    }

  keySerde == null public? int getLastFlushCount(Serde<K>) context.keySerde() {
: keySerde,
          return instanceLastFlushCount.get(      valueSerde == null ? (Serde<V>) context.valueSerde() : valueSerde);
    }

    @Override
    public voidString closename() {
        return innerKeyValueStore.close()name;
    }

    closed = true;@Override
    }

public void init(final ProcessorContext @Overridecontext,
    public boolean persistent() {
        return persistent;
    }

 final StateStore root) @Override{
    public   boolean isOpen() {context.register(root, stateRestoreCallback);
        return !closedinitialized = true;
    }

    publicclosed final= StateRestoreCallbackfalse;
 stateRestoreCallback = new StateRestoreCallback() {}

    @Override
    @Override
  public void flush() {
      public void restore(final byte[] key, instanceLastFlushCount.set(GLOBAL_FLUSH_COUNTER.getAndIncrement());
        wrapped().flush();
        flushed = true;
    }

    public  final byte[] valueint getLastFlushCount() {
         }return instanceLastFlushCount.get();
    };

    @Override
    public void put(final K key, final V valueclose() {
        capturedPutCallswrapped().add(new KeyValue<>(key, value)close();
         innerKeyValueStore.put(key, value)closed = true;
    }

    @Override
    public Vboolean putIfAbsent(final K key, final V value) {persistent() {
        return persistent;
    }

    final@Override
   V originalValuepublic =boolean getisOpen(key); {
        return if (originalValue == null) {!closed;
    }

    public final StateRestoreCallback stateRestoreCallback = new   put(key, value);
StateRestoreCallback() {
        @Override
     capturedPutCalls.add(new KeyValue<>(key, value));   public void restore(final byte[] key,
        }
        return originalValue;
    }

    @Override
    public V delete(final Kbyte[] keyvalue) {
        }
 V value = innerKeyValueStore.delete(key) };

    @Override
    capturedDeleteCalls.add(new public void put(final K key, final V value) {
        capturedPutCalls.add(new KeyValue<>(key, value));
        return valuewrapped().put(keyBytes(key), serdes.rawValue(value));
    }

    @Override
    public voidV putAllputIfAbsent(final List<KeyValue<KK key, V>>final V entriesvalue) {
        for (final KeyValue<K, V> entry : entriesV originalValue = get(key);
        if (originalValue == null) {
            put(entry.key, entry.value);
            capturedPutCalls.add(entrynew KeyValue<>(key, value));
        }
        return originalValue;
    }

    @Override
    public V getdelete(final K key) {
        V value = innerKeyValueStore.get(key)outerValue(wrapped().delete(keyBytes(key)));
        capturedGetCallscapturedDeleteCalls.add(new KeyValue<>(key, value));
        return value;
    }

    @Override
    public KeyValueIterator<K,V>void rangeputAll(final K fromList<KeyValue<K, finalV>> K toentries) {
        returnfor innerKeyValueStore.range(from(final KeyValue<K, to);
V> entry : entries) }{

     @Override
    public KeyValueIterator<K,V>  all() {
put(entry.key, entry.value);
         return innerKeyValueStore  capturedPutCalls.alladd(entry);
        }
    }

    @Override
    public longV approximateNumEntriesget(final K key) {
        V return innerKeyValueStore.approximateNumEntries()value = outerValue(wrapped().get(keyBytes(key)));
        capturedGetCalls.add(new  }
}