import com.microsoft.azure.cosmosdb.ChangeFeedOptions;
import com.microsoft.azure.cosmosdb.ChangeFeedProcessor;
import com.microsoft.azure.cosmosdb.ChangeFeedProcessorBuilder;
import com.microsoft.azure.cosmosdb.ConnectionPolicy;
import com.microsoft.azure.cosmosdb.ConsistencyLevel;
import com.microsoft.azure.cosmosdb.DocumentClient;
import com.microsoft.azure.cosmosdb.DocumentClientException;
import com.microsoft.azure.cosmosdb.DocumentCollection;
import com.microsoft.azure.cosmosdb.FeedOptions;
import com.microsoft.azure.cosmosdb.FeedResponse;
import com.microsoft.azure.cosmosdb.ResourceResponse;
import com.microsoft.azure.cosmosdb.internal.directconnectivity.WFConstants;

import java.util.List;
import java.util.UUID;
import java.util.concurrent.CancellationToken;

public class CosmosDbChangeFeedProcessor {
private static final String DATABASE_ID = "databaseId";
private static final String COLLECTION_ID = "collectionId";
private static final String END_POINT_URI = "https://<your_cosmosdb_account_name>.documents.azure.com:443/";
private static final String PRIMARY_KEY = "<your_primary_key>";

private static void handleChanges(List<Document> docs, CancellationToken cancellationToken) {
// Process the changes
for (Document doc : docs) {
System.out.println("Change: " + doc.toJson());
}
}

public static void main(String[] args) {
// Create a DocumentClient instance
DocumentClient documentClient = new DocumentClient(END_POINT_URI, PRIMARY_KEY,
ConnectionPolicy.GetDefault(), ConsistencyLevel.Session);
try {
  // Get the reference to the database and collection
  DocumentCollection collection = documentClient
      .readCollection(String.format("/dbs/%s/colls/%s", DATABASE_ID, COLLECTION_ID), null)
      .getResource();

  // Define the ChangeFeedOptions
  ChangeFeedOptions changeFeedOptions = new ChangeFeedOptions();
  changeFeedOptions.setPartitionKeyRangeId(WFConstants.NoPartition);
  changeFeedOptions.setStartFromBeginning(true);

  // Define the ChangeFeedProcessor options
  ChangeFeedProcessorOptions changeFeedProcessorOptions = new ChangeFeedProcessorOptions();
  changeFeedProcessorOptions.setLeasePrefix("myChangeFeed");

  // Define the ChangeFeedProcessor
  ChangeFeedProcessor changeFeedProcessor = new ChangeFeedProcessorBuilder()
      .withDocumentClient(documentClient)
      .withFeedCollection(collection)
      .withProcessorOptions(changeFeedProcessorOptions)
      .withHandleChanges(CosmosDbChangeFeedProcessor::handleChanges)
      .build();

  // Start the ChangeFeedProcessor
  changeFeedProcessor.start();
} catch (DocumentClientException e) {
System.err.println("An error occurred while processing the Change Feed: " + e);
    }
  }
}
spam fails_task lang_mismatch pii not_appropriate hate_speech sexual_content quality toxicity humor helpfulness creativity violence