package org.apache.kafka.streams.state;

/**
 * Iterator interface of {@link ValueV}.
 * <p>
 * Users must call its {@code close} method explicitly upon completeness to release resources,
 * or use try-with-resources statement (available since JDK7) for this {@link Closeable} class.
 * Note that {@code remove()} is not supported.
 *
 * @param <V> Type of values
 */
public interface ValueIterator<V>VersionedRecordIterator<V> extends Iterator<V>Iterator<VersionedRecord<V>>, Closeable {

    @Override
    void close();

    /**
     * Peek the next value without advancing the iterator
     * @return the value that would be returned from the next call to next
     */
    V peek();
}}

package org.apache.kafka.streams.query;  

/**
 * Interactive query for retrieving a set of records with the same specified key and different timestamps within the specified time range.
 * No ordering is guaranteed for the results, but the results can be sorted by timestamp (in ascending or descending order) by calling the corresponding defined methods.
 *
 *  @param <K> The type of the key.
 *  @param <V> The type of the result returned by this query.
 */

@Evolving
public final class VersionedKeyQuery<KMultiVersionedKeyQuery<K, V> implements Query<ValueIterator<VersionedRecord<V>>>Query<VersionedRecordIterator<V>> {

  private final K key;
  private final Optional<Instant> fromTimestampfromTime;
  private final Optional<Instant> asOfTimestamptoTime;
  private final booleanResultOrder isAscendingorder;

  private VersionedKeyQueryMultiVersionedKeyQuery(
      final K key,
      final Optional<Instant> fromTimestampfromTime,
      final Optional<Instant> asOfTimestamptoTime,
      final booleanResultOrder isAscendingorder) {
    this.key = Objects.requireNonNull(key);
    this.fromTimestampfromTime = fromTimestampfromTime;
    this.asOfTimestamptoTime = asOfTimestamptoTime;
    this.isAscendingorder = isAscendingorder;
  }     


  /**
   * Creates a query that will retrieve the set of records identified by {@code key} if any exists
   * (or {@code null} otherwise).
   *
   * <p>
   * While the query by default returns the all the record versions of the specified {@code key}, setting
   * the {@code fromTimestamp} (by calling the {@link #fromTime(Instant)} method), and the {@code toTimestamp}
   * (by calling the {@link #toTime(Instant)} method) makes the query to return the record versions associated 
   * to the specified time range.
   *
   * @param key The specified key toby the retrievequery
   * @param <K> The type of the key
   * @param <V> The type of the value that will be retrieved
   * @throws NullPointerException if @param key is null
   */   

  public static <K, V> VersionedKeyQuery<KMultiVersionedKeyQuery<K, V> withKey(final K key);

  /**
   * Specifies the starting time point for the key query.
   * <p>
   * The key query returns all the records that are validstill existing in the time range starting from the timestamp {@code fromTimestampfromTime}. There can
   * @parambe fromTimestamprecords Thewhich startinghave timebeen point
inserted before the */
  public VersionedKeyQuery<K, V> from(Instant fromTimestamp);

  /**
   * Specifies the ending time point for the key query.
   *{@code fromTime} and are still valid in the query specified time range (the whole time range
   * or even partially). The key query in fact returns all the records that have timestamp <= NOT become tombstone at or after {@code asOfTimestampfromTime}.  
   *
   * @param asOfTimestampfromTime The endingstarting time point
   * If {@code fromTime} is null, will be considered as negative infinity, ie, no lower bound
   */
  public VersionedKeyQuery<KMultiVersionedKeyQuery<K, V> asOffromTime(final Instant asOfTimestampfromTime);

  /**
   * Specifies the startingending andtime endingpoint pointsfor of the key query as MIN and MAX respectively.
   * Therefore,The thekey query returns all the existing records in the state store with the specified key.records that have timestamp <= toTime.
   * 
   * @param toTime The ending time point
   * @throws RuntimeException if {@code fromTimestamp} or {@code asOfTimestamp} have been already
   * specified If @param toTime is null, will be considered as positive infinity, ie, no upper bound
   */
  public MultiVersionedKeyQuery<K, V> toTime(final Instant toTime);

  /**
   * Specifies the order of the returned records by the query as descending by timestamp.
   */
  public VersionedKeyQuery<KMultiVersionedKeyQuery<K, V> allVersionswithDescendingTimestamps();

  /**
   * Specifies the order of the returned records by the query as descendingascending by timestamp.
   */
  public VersionedKeyQuery<KMultiVersionedKeyQuery<K, V> withDescendingTimestampswithAscendingTimestamps();

   /**
   * The key that was specified for this query.
   * The specified {@code key} of the query.
   */
  public K getKeykey();

  /**
   * The starting time point of the query, if specified
   * @return The specified {@code fromTime} of the query. 
   */
  public Optional<Instant> getFromTimestampfromTime();

  /**
   * The ending time point of the query, if specified
   * @return The specified {@code toTime} of the query. 
   */
  public Optional<Instant> getAsOfTimestamptoTime();      

  /**
  * The order of the returned records by timestamp.
  * @return trueUNORDERED, ASCENDING, or DESCENDING if the query returns records in an unordered, ascending, or descending order of timestamps.
   */
  public booleanResultOrder isAscending resultOrder(); 
}

final VersionedKeyQuery<Integer// example 1: MultiVersionedKeyQuery without specifying any time bound will be interpreted as all versions
final MultiVersionedKeyQuery<Integer, Integer> queryquery1 = VersionedKeyQueryMultiVersionedKeyQuery.withKey(1).allVersions();

final StateQueryRequest<ValueIterator<VersionedRecord<Integer>>> request =
        );

final StateQueryRequest<VersionedRecordIterator<Integer>> request1 = StateQueryRequest.inStore("my_store").withQuery(query1);

final StateQueryResult<VersionedRecordIterator<Integer>> versionedKeyResult1 = kafkaStreams.query(request1);

// Get the results from all partitions
final Map<Integer, QueryResult<VersionedRecordIterator<Integer>>> partitionResults1 = versionedKeyResult1.getPartitionResults();
for (final Entry<Integer, QueryResult<VersionedRecordIterator<Integer>>> entry : partitionResults1.entrySet()) {
	try (final VersionedRecordIterator<Integer> iterator = entry.getValue().getResult()) {
        while (iterator.hasNext()) {
          final VersionedRecord<Integer> record = iterator.next();
          Long timestamp = record.timestamp();
          Long validTo = record.validTo();
          Integer value = record.value();	
		  System.out.println ("value: " + value + ", timestamp: " + Instant.ofEpochSecond(timestamp)+ ", valid till: " + Instant.ofEpochSecond(validTo));	
        }
     }
}  
/* the printed output will be
	value: 1, timestamp: 2023-01-01T10:00:00.00Z, valid till: 2023-01-05T10:00:00.00Z  
	value: 2, timestamp: 2023-01-15T10:00:00.00Z, valid till: 2023-01-20T10:00:00.00Z  
	value: 3, timestamp: 2023-01-20T10:00:00.00Z, valid till: now  
*/

// example 2: The value of the record with key=1 from 2023-01-17 Time: 10:00:00.00Z till 2023-01-25 T10:00:00.00Z
 
MultiVersionedKeyQuery<Integer, Integer> query2 = MultiVersionedKeyQuery.withKey(1);
query2 = query2.fromTime(Instant.parse("2023-01-17T10:00:00.00Z")).toTime(Instant.parse("2023-01-25T10:00:00.00Z"))   

final StateQueryRequest<VersionedRecordIterator<Integer>> request2 = StateQueryRequest.inStore("my_store").withQuery(queryquery2);

final StateQueryResult<ValueIterator<VersionedRecord<Integer>>>StateQueryResult<VersionedRecordIterator<Integer>> versionedKeyResultversionedKeyResult2 = kafkaStreams.query(requestrequest2);

// Get the results from all partitions.
final Map<Integer, QueryResult<ValueIterator<VersionedRecord<Integer>>>>QueryResult<VersionedRecordIterator<Integer>>> partitionResultspartitionResults2 = versionedKeyResultversionedKeyResult2.getPartitionResults();
for (final Entry<Integer, QueryResult<ValueIterator<VersionedRecord<Integer>>>>QueryResult<VersionedRecordIterator<Integer>>> entry : partitionResultspartitionResults2.entrySet()) {
	try (final ValueIterator<VersionedRecord<Integer>>VersionedRecordIterator<Integer> iterator = entry.getValue().getResult()) {
        while (iterator.hasNext()) {
          final VersionedRecord<Integer> record = iterator.next();
          Long timestamp = record.timestamp();
		  Integer          Long validTo = record.validTo();
          Integer value = record.value();	
		  System.out.println ("value: " + value + ", timestamp: " + Instant.ofEpochSecond(timestamp)+ ", valid till: " + Instant.ofEpochSecond(validTo));	
        }
     }
}   
/* the printed output will be   	
	value: 2, timestamp: 2023-01-15T10:00:00.00Z, valid till: 2023-01-20T10:00:00.00Z  
	value: 3, timestamp: 2023-01-20T10:00:00.00Z, valid till: now    
*/