// Path: src/test/java/com/google/cloud/pubsublite/spark/TestingUtils.java
// public static PslSourceOffset createPslSourceOffset(long... offsets) {
//   Map<Partition, Offset> map = new HashMap<>();
//   int idx = 0;
//   for (long offset : offsets) {
//     map.put(Partition.of(idx++), Offset.of(offset));
//   }
//   return PslSourceOffset.builder().partitionOffsetMap(map).build();
// }
// 
// Path: src/test/java/com/google/cloud/pubsublite/spark/TestingUtils.java
// public static SparkSourceOffset createSparkSourceOffset(long... offsets) {
//   Map<Partition, SparkPartitionOffset> map = new HashMap<>();
//   int idx = 0;
//   for (long offset : offsets) {
//     map.put(Partition.of(idx), SparkPartitionOffset.create(Partition.of(idx), offset));
//     idx++;
//   }
//   return new SparkSourceOffset(map);
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java
// public interface MultiPartitionCommitter extends Closeable {
// 
//   interface CommitterFactory {
//     Committer newCommitter(Partition partition);
//   }
// 
//   void commit(PslSourceOffset offset);
// 
//   void close();
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PartitionSubscriberFactory.java
// public interface PartitionSubscriberFactory extends Serializable {
//   Subscriber newSubscriber(
//       Partition partition, Offset offset, Consumer<List<SequencedMessage>> message_consumer)
//       throws ApiException;
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PerTopicHeadOffsetReader.java
// public interface PerTopicHeadOffsetReader extends Closeable {
// 
//   // Gets the head offsets for all partitions in the topic. Blocks.
//   PslSourceOffset getHeadOffset();
// 
//   @Override
//   void close();
// }


import com.google.cloud.pubsublite.spark.internal.PerTopicHeadOffsetReader;
import com.google.common.collect.ImmutableMap;
import java.util.Optional;
import org.junit.Test;
import static com.google.cloud.pubsublite.spark.TestingUtils.createPslSourceOffset;
import static com.google.cloud.pubsublite.spark.TestingUtils.createSparkSourceOffset;
import static com.google.common.truth.Truth.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.times;
import com.google.api.core.ApiFutures;
import com.google.cloud.pubsublite.Offset;
import com.google.cloud.pubsublite.Partition;
import com.google.cloud.pubsublite.internal.CursorClient;
import com.google.cloud.pubsublite.internal.testing.UnitTestExamples;
import com.google.cloud.pubsublite.spark.internal.MultiPartitionCommitter;
import com.google.cloud.pubsublite.spark.internal.PartitionSubscriberFactory;
/*
 * Copyright 2020 Google LLC
 *
 * 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 com.google.cloud.pubsublite.spark;



public class PslMicroBatchReaderTest {
  private static final PslReadDataSourceOptions OPTIONS =
      PslReadDataSourceOptions.builder()
          .setSubscriptionPath(UnitTestExamples.exampleSubscriptionPath())
          .build();
  private final CursorClient cursorClient = mock(CursorClient.class);
  private final MultiPartitionCommitter committer = mock(MultiPartitionCommitter.class);
  private final PartitionSubscriberFactory partitionSubscriberFactory =
      mock(PartitionSubscriberFactory.class);
  private final PerTopicHeadOffsetReader headOffsetReader = mock(PerTopicHeadOffsetReader.class);
// Path: src/test/java/com/google/cloud/pubsublite/spark/TestingUtils.java
// public static PslSourceOffset createPslSourceOffset(long... offsets) {
//   Map<Partition, Offset> map = new HashMap<>();
//   int idx = 0;
//   for (long offset : offsets) {
//     map.put(Partition.of(idx++), Offset.of(offset));
//   }
//   return PslSourceOffset.builder().partitionOffsetMap(map).build();
// }
// 
// Path: src/test/java/com/google/cloud/pubsublite/spark/TestingUtils.java
// public static SparkSourceOffset createSparkSourceOffset(long... offsets) {
//   Map<Partition, SparkPartitionOffset> map = new HashMap<>();
//   int idx = 0;
//   for (long offset : offsets) {
//     map.put(Partition.of(idx), SparkPartitionOffset.create(Partition.of(idx), offset));
//     idx++;
//   }
//   return new SparkSourceOffset(map);
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/MultiPartitionCommitter.java
// public interface MultiPartitionCommitter extends Closeable {
// 
//   interface CommitterFactory {
//     Committer newCommitter(Partition partition);
//   }
// 
//   void commit(PslSourceOffset offset);
// 
//   void close();
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PartitionSubscriberFactory.java
// public interface PartitionSubscriberFactory extends Serializable {
//   Subscriber newSubscriber(
//       Partition partition, Offset offset, Consumer<List<SequencedMessage>> message_consumer)
//       throws ApiException;
// }
// 
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PerTopicHeadOffsetReader.java
// public interface PerTopicHeadOffsetReader extends Closeable {
// 
//   // Gets the head offsets for all partitions in the topic. Blocks.
//   PslSourceOffset getHeadOffset();
// 
//   @Override
//   void close();
// }


// Path: src/test/java/com/google/cloud/pubsublite/spark/PslMicroBatchReaderTest.java
import com.google.cloud.pubsublite.spark.internal.PerTopicHeadOffsetReader;
import com.google.common.collect.ImmutableMap;
import java.util.Optional;
import org.junit.Test;
import static com.google.cloud.pubsublite.spark.TestingUtils.createPslSourceOffset;
import static com.google.cloud.pubsublite.spark.TestingUtils.createSparkSourceOffset;
import static com.google.common.truth.Truth.assertThat;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.*;
import static org.mockito.Mockito.times;
import com.google.api.core.ApiFutures;
import com.google.cloud.pubsublite.Offset;
import com.google.cloud.pubsublite.Partition;
import com.google.cloud.pubsublite.internal.CursorClient;
import com.google.cloud.pubsublite.internal.testing.UnitTestExamples;
import com.google.cloud.pubsublite.spark.internal.MultiPartitionCommitter;
import com.google.cloud.pubsublite.spark.internal.PartitionSubscriberFactory;

/*
 * Copyright 2020 Google LLC
 *
 * 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 com.google.cloud.pubsublite.spark;



public class PslMicroBatchReaderTest {
  private static final PslReadDataSourceOptions OPTIONS =
      PslReadDataSourceOptions.builder()
          .setSubscriptionPath(UnitTestExamples.exampleSubscriptionPath())
          .build();
  private final CursorClient cursorClient = mock(CursorClient.class);
  private final MultiPartitionCommitter committer = mock(MultiPartitionCommitter.class);
  private final PartitionSubscriberFactory partitionSubscriberFactory =
      mock(PartitionSubscriberFactory.class);
  private final PerTopicHeadOffsetReader headOffsetReader = mock(PerTopicHeadOffsetReader.class);
  private static final long MAX_MESSAGES_PER_BATCH = 20000;