Java sample to backfill meters from MongoDB

this java sample demonstrates how to backfill historical meter data stored in mongodb using the amberflo java sdk the sample will perform the following tasks reads entries from a mongodb collection , where each entry represents a usage event for each entry ingests the event into amberflo as a metered usage record uses the entry's creation timestamp (in utc) as the meter event time assigns the entry’s unique id as the uniqueid for the meter event to prevent duplication ensure idempotency in case the script is re run this setup ensures reliable, repeatable ingestion of historical data while maintaining data integrity and preventing duplicate usage records in amberflo maven dependencies (check for latest versions) \<dependencies> \<dependency> \<groupid>io amberflo\</groupid> \<artifactid>metering java client\</artifactid> \<version>1 3 6\</version> \</dependency> \<dependency> \<groupid>org mongodb\</groupid> \<artifactid>mongo java driver\</artifactid> \<version>3 5 0\</version> \</dependency> \<dependency> \<groupid>com stripe\</groupid> \<artifactid>stripe java\</artifactid> \<version>20 52 0\</version> \</dependency> \</dependencies> package com amberflo backfill; import com amberflo metering ingest meteringcontext; import com amberflo metering ingest meter message domain; import com amberflo metering ingest meter message metermessage; import com amberflo metering ingest meter message metermessagebuilder; import com amberflo metering ingest meter message region; import com mongodb mongoclient; import com mongodb mongoclienturi; import com mongodb client finditerable; import com mongodb client mongocollection; import com mongodb client mongocursor; import com mongodb client mongodatabase; import org bson document; import java time instant; import java time localdatetime; import java time zoneid; import java util hashmap; import java util map; import static com amberflo metering ingest meteringcontext metering; public class backfillmongodbdata { //todo set the database connection, queries, and amberflo info private static final string mongo host = "sample db mongo host"; private static final int mongo port = 27017; private static final string mongo user = "sample user"; private static final string mongo pass = "pass"; private static final string db name = "sampledb"; private final static string amberflo api key = "amberflo api key"; //end todo private final static string event create = "event create"; public static void main(final string\[] args) throws exception { meteringcontext createorreplacecontext( amberflo api key, "events api", domain prod, region us west, 1, 10 ); // mongodb connection string final string connectionstring = string format("mongodb //%s %s@%s %s", mongo user, mongo pass, mongo host, mongo port); final mongoclienturi uri = new mongoclienturi(connectionstring); // connecting to the mongodb server using the given client uri final mongoclient mongoclient = new mongoclient(uri); // fetching the database from the mongodb final mongodatabase db = mongoclient getdatabase(db name); // fetching the collection from the mongodb final mongocollection\<document> col = db getcollection("events"); // performing a read operation on the collection final finditerable\<document> fi = col find(); final mongocursor\<document> cursor = fi iterator(); int counter = 1; try { while (cursor hasnext()) { final document document = cursor next(); if (document == null) { continue; } //example of a nested document final document eventdata = (document) document get("eventdata"); final string eventsourcetype = eventdata get("event source type") tostring(); final string eventtype = eventdata get("event type") tostring(); final string metername = event create; final string customerid = eventdata get("customer id") tostring(); final string id = document get(" id") tostring(); //converting timestamp stored as double epoch final long eventtimeepoch = ((double) double parsedouble(document get("created time epoch utc") tostring())) longvalue(); final localdatetime eventtime = instant ofepochmilli(eventtimeepoch) atzone(zoneid of("utc")) tolocaldatetime(); system out println(string format("ingesting event source type=%s, customerid=%s, id=%s, eventtime=%s, eventtype=%s, metername %s", eventsourcetype, customerid, id, eventtime, eventtype, metername)); system out println(document tojson()); final map\<string, string> dimensions = new hashmap(); dimensions put("event source type", eventsourcetype); dimensions put("event type", eventtype); final metermessage metermessage = metermessagebuilder createinstance(metername, eventtime, customerid) setmetervalue(1) setuniqueid(id) setdimensionsmap(dimensions) build(); metering() meter(metermessage); if (counter % 100 == 0) { thread sleep(5000); } counter++; } system out println("total meters ingested " + counter); } finally { cursor close(); mongoclient close(); meteringcontext flushandclose(); system out println("done!"); } } }