/**
     * Flush any cached data
     * @deprecated Use {@link #commit(Map)} instead.
     */
    @Deprecated
    default void flush() {
        commit(Collections.emptyMap());
    }

    /**
     * Commits any uncommitted records to this StateStore with the given {@code changelogOffsets}.
     * <p>
     * If this StateStore has any open {@link Transaction transactions}, they will be {@link Transaction#commit(Map)
     * committed}.
     * <p>
     * The given {@code changelogOffsets} will be committed to the StateStore along with the uncommitted {@link
     * Transaction transactions}. If this StateStore supports atomic transactions, offsets are guaranteed to be
     * committed atomically with the records they correspond to.
     *
     * @param changelogOffsets The input/changelog topic offsets of the records being committed.
     */
    @Evolving
    default void commit(final Map<TopicPartition, Long> changelogOffsets) {
        flush();
    }

    /**
     * Returns whether this StateStore manages its own changelog offsets.
     * <p>
     * This can only return {@code true} if {@link #persistent()} also returns {@code true}, as non-persistent stores
     * have no changelog offsets to manage.
     *
     * @return Whether this StateStore manages its own changelog offsets.
     */
    @Evolving
    default boolean managesOffsets() {
        return false;
    }

    /**
     * Returns the committed offset for the given partition.
     * <p>
     * If this store is not {@link #persistent()} or does not {@link #managesOffsets() manage its own offsets},
     * it will always yield {@code null}.
     * <p>
     * If {@code topicPartition} is a changelog partition or Topology input partition for this StateStore, this method
     * will return the committed offset for that partition.
     * <p>
     * If no committed offset exists for the given partition, or if the partition is not a changelog or input partition
     * for the store, {@code null} will be returned.
     *
     * @param topicPartition The changelog/input partition to get the committed offset for.
     * @return The committed offset for the given partition, or {@code null} if no committed offset exists, or if this
     *         store does not contain committed offsets.
     */
    @Evolving
    default Long getCommittedOffset(final TopicPartition topicPartition) {
        return null;
    }

    /**
     * Creates a new {@link Transaction} for reading/writing to this state store.
     * <p>
     * State stores that do not support transactions will return {@code this} instead, which should be considered a
     * transaction that doesn't provide any isolation or atomicity guarantees.
     * <p>
     * {@link Transaction Transactions} are <em>not thread-safe</em>, and should not be shared among threads. New
     * threads should use this method to create a new transaction, instead of sharing an existing one.
     * <p>
     * Transactions crated by this method should have the same {@link IsolationLevel} as the {@link StateStoreContext}
     * that this store was {@link #init(StateStoreContext, StateStore) initialized with}.
     * <p>
     * To avoid resource leaks, it is recommended to extend {@link AbstractTransactionalStore}, which will track open
     * transactions and automatically close all open transactions when the store is closed.
     *
     * @return A new {@link Transaction} to isolate reads/writes to this {@link StateStore}. The Transaction
     *         <strong>MUST</strong> be {@link Transaction#commit(Map) committed} or {@link Transaction#close() closed}
     *         when you are finished with it, to prevent resource leaks.
     */
    @Evolving
    default StateStore newTransaction() {
        return this;
    }

    /**
     * Return an approximate count of records not yet committed to this StateStore.
     * <p>
     * This method will return an approximation of the number of records that would be committed by the next call to
     * {@link #commit(Map)}.
     * <p>
     * If this StateStore is unable to approximately count uncommitted records, it will return {@code -1}.
     * If this StateStore does not support atomic transactions, it will return {@code 0}, because records will always be
     * immediately written to a non-transactional store, so there will be none awaiting a {@link #commit(Map)}.
     *
     * @return The approximate number of records awaiting {@link #commit(Map)}, {@code -1} if the number of
     *         uncommitted records can't be counted, or {@code 0} if this StateStore does not support transactions.
     */
    @Evolving
    default long approximateNumUncommittedEntries() {
        return 0;
    }

   /**
     * Return an approximate count of memory used by records not yet committed to this StateStore.
     * <p>
     * This method will return an approximation of the memory would be freed by the next call to {@link #commit(Map)}.
     * <p>
     * If this StateStore is unable to approximately count uncommitted memory usage, it will return {@code -1}.
     * If this StateStore does not support atomic transactions, it will return {@code 0}, because records will always be
     * immediately written to a non-transactional store, so there will be none awaiting a {@link #commit(Map)}.
     *
     * @return The approximate size of all records awaiting {@link #commit(Map)}, {@code -1} if the size of uncommitted
     *         records can't be counted, or {@code 0} if this StateStore does not support transactions.
     */
    @Evolving
    default long approximateNumUncommittedBytes() {
        return 0;
    }

    /**
     * Returns the {@link IsolationLevel} that every {@link Transaction} created by {@link StateStore#newTransaction()}
     * should use.
     * <p>
     * The default implementation of this method will use {@link IsolationLevel#READ_COMMITTED READ_COMMITTED} if the
     * app is {@link #appConfigs() configured} to use an {@link StreamsConfig#EXACTLY_ONCE_V2 exactly-once} {@link
     * StreamsConfig#PROCESSING_GUARANTEE_CONFIG processing guarantee}. Otherwise, it will be {@link
     * IsolationLevel#READ_UNCOMMITTED READ_UNCOMMITTED}.
     *
     * @return The isolation level for every transaction created by state stores for this context.
     */
    default IsolationLevel isolationLevel() {
        return StreamsConfigUtils.eosEnabled(new StreamsConfig(appConfigs())) ?
                IsolationLevel.READ_COMMITTED : IsolationLevel.READ_UNCOMMITTED;
    }

/**
 * Represents a read/write transaction on a state store.
 * <p>
 * For compatibility, transactions implement the entire {@link StateStore} interface, however, they only represent a
 * <em>view</em> on an underlying {@link StateStore}; therefore methods that explicitly act on the entire store, such
 * as {@link #init} will do nothing, and others like {@link #commit()} and {@link #close()} will act upon this
 * transaction only.
 * <p>
 * Transactions are <em>NOT</em> thread-safe, and they should not be shared among multiple threads. Threads should
 * create their own {@link StateStore#newTransaction() new transaction}, which will ensure proper isolation of
 * concurrent changes.
 * <p>
 * For resource-safety, Transactions implement the {@link AutoCloseable} interface, enabling try-with-resources:
 * <pre>
 *     try (final Transaction transaction = store.newTransaction()) {
 *         transaction.put("foo", "bar");
 *         transaction.commit(Map);
 *     }
 * </pre>
 * If you are not using try-with-resources, you <em>must</em> call either {@link #commit(Map)} exactly-once, or {@link
 * #close()} at least once, or risk possible resource leaks.
 */
public interface Transaction extends StateStore, AutoCloseable {

    /**
     * The {@link IsolationLevel} that reads and writes in this transaction are subject to.
     * <p>
     * In the context of a {@link org.apache.kafka.streams.processor.StateStore} transaction, these Isolation Levels
     * adhere to the <a href="https://en.wikipedia.org/wiki/Isolation_(database_systems)#Read_phenomena">ANSI SQL 92
     * definitions</a>.
     * <p>
     * All isolation levels guarantee "read-your-own-writes", i.e. that writes in the transaction will be seen by
     * subsequent reads <em>from within the same transaction</em>. Other guarantees vary by isolation level:
     * <p>
     * <table>
     *     <tr>
     *         <th>Isolation Level</th>
     *         <th>Description</th>
     *         <th>Permitted Read Phenomena</th>
     *     </tr>
     *     <tr>
     *         <td>{@link IsolationLevel#READ_UNCOMMITTED}</td> // todo: ALOS
     *         <td>Allows queries to read writes from all ongoing transactions that have not-yet been committed.</td>
     *         <td>dirty reads, non-repeatable reads, phantom reads</td>
     *     </tr>
     *     <tr>
     *         <td>{@link IsolationLevel#READ_COMMITTED}</td> // todo: EOS
     *         <td>Allows queries to only read writes that have been committed to the StateStore. Writes by an ongoing
     *         transaction are not visible <em>until that transaction commits</em>.</td>
     *         <td>non-repeatable reads, phantom reads</td>
     *     </tr>
     * </table>
     * <p>
     * Under {@link IsolationLevel#READ_UNCOMMITTED}, there are no guarantees on when records from other transactions
	 * become visible, therefore implementations <em>may</em> refrain from making uncommitted records visible to other
     * transactions until they're committed, if doing so would improve performance.
     */
    IsolationLevel isolationLevel();

    /**
     * Initializes this Transaction.
     * <p>
     * Most transactions require no explicit initialization, therefore the default implementation of this method does
     * nothing.
     *
     * @deprecated Since 2.7.0, the parent method has been deprecated in favor of {@link
     *             #init(StateStoreContext, StateStore)}. Use that method instead.
     */
    @Override @Deprecated
    default void init(final ProcessorContext context, final StateStore root) {
    }

    /**
     * Initializes this Transaction.
     * <p>
     * Most transactions require no explicit initialization, therefore the default implementation of this method does
     * nothing.
     */
    @Override
    default void init(final StateStoreContext context, final StateStore root) {
    }

    /**
     * Is this transaction is open for reading and writing.
     * @return {@code true} if this Transaction can be read from/written to, or {@code false} if it is no longer
     *         accessible.
     */
    @Override
    boolean isOpen();

    /**
     * Creates a new {@link Transaction} from an existing transaction.
     * <p>
     * This enables potentially re-using resources from an existing, no longer in-use transaction, instead of creating
     * a new one.
     * <p>
     * This method should only be called if the current transaction is guaranteed to no longer be in-use.
     * Implementations may return either a new Transaction instance, or a reference to themselves, only if they are able
     * to reset to being available for use.
     * <p>
     * {@link Transaction Transactions} are <em>not thread-safe</em>, and should not be shared among threads. New
     * threads should use this method to create a new transaction, instead of sharing an existing one.
     * <p>
     * Transactions created by this method will have the same {@link IsolationLevel} as the {@link StateStoreContext}
     * that this transaction was {@link #init(StateStoreContext, StateStore) initialized with}.
     *
     * @return A new {@link Transaction} to control reads/writes to the same {@link StateStore}. The Transaction
     *         <em>MUST</em> be {@link Transaction#commit(Map) committed} or {@link Transaction#close() closed} when you
     *         are finished with it, to prevent resource leaks.
     */
    @Override
    StateStore newTransaction();

    /**
     * Commit and close this Transaction.
     * <p>
     * Any records held by this Transaction will be written to the underlying state store.
     * <p>
     * Once this method returns successfully, this Transaction will no longer be available to use for reads/writes.
     *
     * @see #close() to close the Transaction without writing data to the underlying state store.
     */
    @Override
    void commit(final Map<TopicPartition, Long> changelogOffsets);

    /**
     * Closes this Transaction, without committing records.
     * <p>
     * Any uncommitted records will <em>not</em> be written to the underlying state store, and will instead be
     * discarded.
     * <p>
     * This method should be used to "rollback" a Transaction that should not be {@link #commit(Map) committed}.
     * <p>
     * The underlying {@link StateStore} will <em>not</em> be closed by this method.
     * <p>
     * This method is idempotent: repeatedly calling {@code close()} will not produce an error.
     *
     * @see #commit(Map) to close this Transaction by writing its records to the underlying state store.
     */
    @Override
    void close();
}

/**
 * Base implementation of {@link Transaction transactions} created by an implementation of {@link
 * AbstractTransactionalStore}.
 * <p>
 * This base implementation provides the following functionality:
 * <ul>
 *     <li>Registration of callbacks that are invoked after this Transaction has closed. Used by {@link
 *     AbstractTransactionalStore} to track open transactions that need to be closed when the parent store closes.</li>
 *     <li>Tracking/delegation of {@link #isOpen()}, {@link #persistent()}, {@link #name()} and {@link
 *     #isolationLevel()} methods.</li>
 *     <li>Provides a {@link #validateIsOpen()} method, useful for ensuring the Transaction state in query methods.</li>
 *     <li>Provides base implementations of {@link #commit(Map)} and {@link #close()} that properly handle idempotence and
 *     invoking any registered callbacks.</li>
 * </ul>
 * <p>Note: none of the above methods are marked {@code final}, to enable power-users to provide alternatives. However,
 * this should only be done with great care, and only if absolutely necessary.</p>

 *
 * @param <S> The type of the {@link StateStore} that spawned this transaction.
 */
public abstract class AbstractTransaction<S extends StateStore> implements Transaction {

    /**
     * Commit and close this Transaction.
     * 
     * @see Transaction#commit(Map)
     */
    public abstract void commitTransaction(final Map<TopicPartition, Long> offsets);

    /**
	 * Closes this Transaction, without committing records.
     *
     * @see Transaction#close()
     */
    public abstract void closeTransaction();
}

/**
 * Base class for transactional stores, that tracks open transactions and can close them all.
 * <p>
 * Transactions created using {@link #newTransaction()} will be automatically tracked and closed when this store is
 * closed.
 *
 * @see Transaction
 * @see AbstractTransaction
 */
public abstract class AbstractTransactionalStore implements StateStore {

    /**
     * Creates a new {@link Transaction}.
     * <p>
     * Implementations of this class should implement this method, instead of {@link #newTransaction()}, which
     * will call this method to produce the new transaction.
     * <p>
     * Transactions produced by this method must be an instance of {@link AbstractTransaction}, to enable the
     * automatic management of transaction resources.
     * 
     * @return A new {@link Transaction}.
     */
    public abstract AbstractTransaction<? extends AbstractTransactionalStore> createTransaction();
}