public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
            var batch = new TableBatchOperation();

            foreach(var msg in messages)
            {
                var snap = JsonConvert.DeserializeObject<BusSnapshotInfo>(Encoding.UTF8.GetString(msg.GetBytes()));

                var entity = new DynamicTableEntity(snap.RouteShortName, snap.VehicleId.ToString());

                entity.Properties.Add("RouteShortName", EntityProperty.GeneratePropertyForString(snap.RouteShortName));
                entity.Properties.Add("VehicleId", EntityProperty.GeneratePropertyForInt(snap.VehicleId));
                entity.Properties.Add("TripId", EntityProperty.GeneratePropertyForInt(snap.TripId));
                entity.Properties.Add("Latitude", EntityProperty.GeneratePropertyForDouble(snap.Latitude));
                entity.Properties.Add("Longitude", EntityProperty.GeneratePropertyForDouble(snap.Longitude));
                entity.Properties.Add("DirectionOfTravel", EntityProperty.GeneratePropertyForString(snap.DirectionOfTravel.ToString()));
                entity.Properties.Add("NextStopId", EntityProperty.GeneratePropertyForInt(snap.NextStopId));
                entity.Properties.Add("Timeliness", EntityProperty.GeneratePropertyForString(snap.Timeliness.ToString()));
                entity.Properties.Add("TimelinessOffset", EntityProperty.GeneratePropertyForInt(snap.TimelinessOffset));
                entity.Properties.Add("Timestamp", EntityProperty.GeneratePropertyForDateTimeOffset(snap.Timestamp));

                batch.Add(TableOperation.InsertOrReplace(entity));
            }

            var tableClient = _account.CreateCloudTableClient();

            var table = tableClient.GetTableReference("snapshots");

            await table.CreateIfNotExistsAsync();

            await table.ExecuteBatchAsync(batch);

            await context.CheckpointAsync();
        }

        void ClearTable(CloudTable table)
        {
            var deviceIds = _deviceService.GetDeviceIds();

            foreach (var partitionKey in deviceIds)
            {
                TableBatchOperation batchDelete = new TableBatchOperation();

                // gets all the entities in the table for this partition key
                string partitionCondition = TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey);
                List<DynamicTableEntity> entities = table.ExecuteQuery(new TableQuery().Where(partitionCondition)).ToList();

                entities.ForEach(e =>
                {
                    batchDelete.Add(TableOperation.Delete(e));

                    // Azure has a limit on batch operations
                    if (batchDelete.Count == 100)
                    {
                        table.ExecuteBatch(batchDelete);
                        batchDelete = new TableBatchOperation();
                    }
                });

                // flush out whatever is left
                if (batchDelete.Count > 0)
                {
                    table.ExecuteBatch(batchDelete);
                }
            }
        }

        public void Insert()
        {
            TableBatchOperation batchOperation = new TableBatchOperation();
            Hashtable barcode = new Hashtable();
            StringBuilder sb = new StringBuilder();
            for (int i = id; i < id + 100 && i < Program._DATA_TABLE.Rows.Count; i++)
            {
                if (!barcode.ContainsKey(Program._DATA_TABLE.Rows[i]["Barcode"].ToString().Trim()))
                {
                    try
                    {
                        sb.Append(Program._DATA_TABLE.Rows[i]["Barcode"].ToString().Trim() + "|");
                        ShopBarcodeEntity data = new ShopBarcodeEntity(Program._DATA_TABLE.Rows[i]["Shop"].ToString().Trim(), Program._DATA_TABLE.Rows[i]["Barcode"].ToString().Trim());
                        data.OrderNo = Program._DATA_TABLE.Rows[i]["BillNumber"].ToString().Trim();
                        data.Product = Program._DATA_TABLE.Rows[i]["Product"].ToString().Trim().PadLeft(8, '0');
                        data.Cost = double.Parse(Program._DATA_TABLE.Rows[i]["SellPrice"].ToString().Trim());
                        data.SellFinished = false;
                        batchOperation.InsertOrMerge(data);
                        barcode[Program._DATA_TABLE.Rows[i]["Barcode"].ToString().Trim()] = true;
                    }
                    catch { }
                }
            }

            try
            {
                Program._RECORD++;
                Console.WriteLine("Insert Record {0}-{1}\t\tTotal {2} Records", id + 1, id + 100, Program._RECORD*100);
                Program._CLOUD_TABLE.ExecuteBatch(batchOperation);
            }
            catch (Exception e)
            {
                Program.WriteErrorLog("Record " + (id + 1) + "-" + (id + 100) + " Error \n" + sb.ToString() + "\n" + e.Message + "\n" + e.StackTrace);
            }
        }

        public async void Add(IList<ITableEntity> notes)
        {
            var batchOperation = new TableBatchOperation();

            notes.ToList().ForEach(n => batchOperation.Insert(n));
            await _table.ExecuteBatchAsync(batchOperation);
        }

        public void TableBatchAddNullShouldThrow()
        {
            TableBatchOperation batch = new TableBatchOperation();
            try
            {
                batch.Add(null);
                Assert.Fail();
            }
            catch (ArgumentNullException)
            {
                // no op
            }
            catch (Exception)
            {
                Assert.Fail();
            }

            try
            {
                batch.Insert(0, null);
                Assert.Fail();
            }
            catch (ArgumentNullException)
            {
                // no op
            }
            catch (Exception)
            {
                Assert.Fail();
            }
        }

        public long ClearTable()
        {
            long deletionCount = 0;
            // Construct the query operation for all customer entities where PartitionKey="Smith".
            var list = new List<string>();
            list.Add("PartitionKey");
            list.Add("RowKey");
            TableQuery<SensorValueEntity> query = new TableQuery<SensorValueEntity>().Select(list).Take(100);
            var results = table.ExecuteQuery(query);

            if (results.Count() < 1)
                return deletionCount;
            foreach(var resultGroup in results.GroupBy(a => a.PartitionKey))
            {
                TableBatchOperation batchOperation = new TableBatchOperation();
                foreach (var result in resultGroup)
                {
                    batchOperation.Delete(result);
                    deletionCount++;
                }
                table.ExecuteBatch(batchOperation);
            }

            return deletionCount;
        }

        public List<Reply> GetReplyNotif(string userid)
        {
            TableQuery<ReplyNotificationEntifity> query = new TableQuery<ReplyNotificationEntifity>().Where(TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, userid));

            List<Reply> replies = new List<Reply>();
            TableBatchOperation batchOperation = new TableBatchOperation();
            int count = 0;

            // Print the fields for each customer.
            foreach (ReplyNotificationEntifity entity in _replyNotification.ExecuteQuery(query))
            {
                replies.Add(JsonConvert.DeserializeObject<Reply>(entity.Content));

                batchOperation.Add(TableOperation.Delete(entity));
                count++;

                if((count % 100) == 0){
                    _replyNotification.ExecuteBatch(batchOperation);
                    batchOperation = new TableBatchOperation();
                    count = 0;
                }
            }

            if (count > 0)
            {
                _replyNotification.ExecuteBatch(batchOperation);
            }

            return replies;
        }

        private void InsertBets()
        {
            if (this.Bets.Count == 0)
            {
                return;
            }

            var tableBets = this.CreateTableClient("Bet");

            tableBets.CreateIfNotExists();

            var batchOperation = new TableBatchOperation();

            foreach (var kvp in this.Bets)
            {
                var bet = new BetEntity();
                bet.TicketNumber = kvp.Key;
                bet.BetNumber = kvp.Value;
                bet.RaffleId = this.RaffleId;

                batchOperation.Insert(bet);
            }

            tableBets.ExecuteBatch(batchOperation);
        }

        async Task<CommandResult> DoInsert(CloudTable table, long n, Func<long, EntityNk[]> entityFactory)
        {
            
            var batchOperation = new TableBatchOperation();

            foreach (var e in entityFactory(n))
            {
                batchOperation.Insert(e);
            }

            var cresult = new CommandResult { Start = DateTime.UtcNow.Ticks };
            var cbt = 0L;
            var context = GetOperationContext((t) => cbt = t);
            try
            {
                var results = await table.ExecuteBatchAsync(batchOperation, operationContext: context);
                cresult.Elapsed = cbt;
            }
            catch (Exception ex)
            {
                cresult.Elapsed = -1;
                Console.Error.WriteLine("Error DoInsert {0} {1}", n, ex.ToString());
            }
            return cresult;
        }

        public static bool InsertEntityToAzureTable(string tableName, List<CustomerEntity> entityList)
        {
            System.Diagnostics.Trace.TraceInformation(string.Format("Date = {0}, AzureTableOperation.InsertEntityToAzureTable2 is Called", DateTime.Now));
            // Retrieve the storage account from the connection string.
            CloudStorageAccount storageAccount = CloudStorageAccount.Parse(GetStorageConnectionString());

            // Create the table client.
            CloudTableClient tableClient = storageAccount.CreateCloudTableClient();

            // Create the CloudTable object that represents the "people" table.
            CloudTable table = tableClient.GetTableReference(tableName);

            // Create the batch operation.
            TableBatchOperation batchOperation = new TableBatchOperation();

            // Add both customer entities to the batch insert operation.
            foreach (CustomerEntity entity in entityList)
            {
                batchOperation.Insert(entity);
            }

            // Execute the insert operation.
            IList<TableResult> resultList = table.ExecuteBatch(batchOperation);
            System.Diagnostics.Trace.TraceInformation(string.Format("Date = {0}, AzureTableOperation.InsertEntityToAzureTable2 insert entity to {1}: {2}", DateTime.Now, tableName, resultList.Count));

            //TODO: how to determine whether the operation is successful
            return true;
        }

 public ActionResult DeleteConfirmed(string partitionKey)
 {
     // Delete all rows for this mailing list, that is,
     // Subscriber rows as well as MailingList rows.
     // Therefore, no need to specify row key.
     var query = new TableQuery<MailingList>().Where(TableQuery.GenerateFilterCondition("PartitionKey", QueryComparisons.Equal, partitionKey));
     var listRows = mailingListTable.ExecuteQuery(query).ToList();
     var batchOperation = new TableBatchOperation();
     int itemsInBatch = 0;
     foreach (MailingList listRow in listRows)
     {
         batchOperation.Delete(listRow);
         itemsInBatch++;
         if (itemsInBatch == 100)
         {
             mailingListTable.ExecuteBatch(batchOperation);
             itemsInBatch = 0;
             batchOperation = new TableBatchOperation();
         }
     }
     if (itemsInBatch > 0)
     {
         mailingListTable.ExecuteBatch(batchOperation);
     }
     return RedirectToAction("Index");
 }

        public async Task Persist(ProjectionMetaData metadata)
        {
            Trace.TraceInformation("Preparing batch for projection {0}.", metadata.ProjectionType);

            var batch = new TableBatchOperation();

           
                var entity = new DictionaryTableEntity
                {
                    PartitionKey = metadata.ProjectionType,
                    RowKey = Handler.Current()
                };

                entity.Add("ProjectionHash", metadata.ProjectionHash);
                batch.Add(TableOperation.InsertOrReplace(entity));


                Trace.TraceInformation("Executing batch for projection {0}.", metadata.ProjectionType);
                await _table.ExecuteBatchAsync(batch).ContinueWith(r =>
                {
                    Trace.TraceInformation("Batch for projection {0} complete {1} exceptions.", metadata.ProjectionType, r.Exception != null ? "with" : "without");
                    if (r.Exception != null)
                    {
                        r.Exception.Handle(exception =>
                        {
                            Trace.TraceError(exception.ToString());
                            return true;
                        });
                    }
                });
        }

 public AzureTableContext(CloudTable table, IEncryptor encryptor)
 {
     _table = table;
     _encryptor = encryptor;
     _context = new TableBatchOperation();
     _deleteBackupContext = new TableBatchOperation();
 }

        public void Insert()
        {
            TableBatchOperation batchOperation = new TableBatchOperation();
            Hashtable bank = new Hashtable();

            for (int i = id; i < id + 100 && i < Program._DATA_TABLE.Rows.Count; i++)
            {
                if (!bank.ContainsKey(Program._DATA_TABLE.Rows[i]["ID"].ToString().Trim()))
                {
                    try
                    {
                        DynamicTableEntity data = new DynamicTableEntity();
                        data.PartitionKey = "88888888";
                        data.RowKey = Program._DATA_TABLE.Rows[i]["ID"].ToString().Trim().PadLeft(6, '0');
                        Dictionary<string, EntityProperty> properties = new Dictionary<string, EntityProperty>();

                        properties.Add("Bank", new EntityProperty(Program._DATA_TABLE.Rows[i]["Bank"].ToString().Trim()));
                        properties.Add("BranchName", new EntityProperty(Program._DATA_TABLE.Rows[i]["BranchName"].ToString().ToLower().Trim()));
                        properties.Add("AccountNumber", new EntityProperty(Program._DATA_TABLE.Rows[i]["AccountNumber"].ToString().Trim()));
                        properties.Add("AccountName", new EntityProperty(Program._DATA_TABLE.Rows[i]["AccountName"].ToString().Trim()));
                        properties.Add("AccountType", new EntityProperty(int.Parse(Program._DATA_TABLE.Rows[i]["AccountType"].ToString().Trim())));

                        //BankEntity data = new BankEntity("88888888", Program._DATA_TABLE.Rows[i]["ID"].ToString().Trim().PadLeft(6, '0'));
                        //data.Bank = Program._DATA_TABLE.Rows[i]["Bank"].ToString().Trim();
                        //data.BranchName = Program._DATA_TABLE.Rows[i]["BranchName"].ToString().ToLower().Trim();
                        //data.AccountNumber = Program._DATA_TABLE.Rows[i]["AccountNumber"].ToString().Trim();
                        //data.AccountName = Program._DATA_TABLE.Rows[i]["AccountName"].ToString().Trim();
                        //data.AccountType = int.Parse(Program._DATA_TABLE.Rows[i]["AccountType"].ToString().Trim());
                        batchOperation.InsertOrMerge(data);
                        recBank++;
                        bank[Program._DATA_TABLE.Rows[i]["ID"].ToString().Trim()] = true;
                    }
                    catch { }
                }
            }

            try
            {
                if (Program._DATA_TABLE.Rows.Count > 0)
                {
                    if (Program._UPDATE)
                    {
                        Program._RECORD++;
                        Console.WriteLine("Update Record {0}-{1}\t\tTotal {2} Records", id + 1, id + 100, Program._RECORD * 100);
                        Program._CLOUD_TABLE.ExecuteBatch(batchOperation);
                    }
                    else
                    {
                        Program._RECORD++;
                        Console.WriteLine("Insert Record {0}-{1}\t\tTotal {2} Records", id + 1, id + 100, Program._RECORD * 100);
                        Program._CLOUD_TABLE.ExecuteBatch(batchOperation);
                    }
                }

            }
            catch (Exception e)
            {
                Program.WriteErrorLog("Record " + (id + 1) + "-" + (id + 100) + " Error \n" + e.Message + "\n" + e.StackTrace);
            }
        }

        /// <summary>
        /// Emit a batch of log events, running to completion synchronously.
        /// </summary>
        /// <param name="events">The events to emit.</param>
        /// <remarks>Override either <see cref="PeriodicBatchingSink.EmitBatch"/> or <see cref="PeriodicBatchingSink.EmitBatchAsync"/>,
        /// not both.</remarks>
        protected override void EmitBatch(IEnumerable<LogEvent> events)
        {
            var operation = new TableBatchOperation();
            
            var first = true;
            
            foreach (var logEvent in events)
            {
                if (first)
                {
                    //check to make sure the partition key is not the same as the previous batch
                    if (partitionKey != logEvent.Timestamp.Ticks)
                    {
                        batchRowId = 0; //the partitionkey has been reset
                        partitionKey = logEvent.Timestamp.Ticks; //store the new partition key
                    }
                    first = false;
                }

                var logEventEntity = new LogEventEntity(logEvent, _formatProvider, partitionKey);
                logEventEntity.RowKey += "|" + batchRowId;
                operation.Add(TableOperation.Insert(logEventEntity));

                batchRowId++;
            }
            _table.ExecuteBatch(operation);
        }

        static void CreateCustomerMetadata(CloudTableClient tableClient)
        {
            Console.WriteLine("Creating customers metadata...");

            CloudTable customersMetadataTable = tableClient.GetTableReference("customersmetadata");
            customersMetadataTable.CreateIfNotExists();

            var msftAddress1 = new DictionaryTableEntity();
            msftAddress1.PartitionKey = "MSFT";
            msftAddress1.RowKey = "ADDRESS-" + Guid.NewGuid().ToString("N").ToUpper();
            msftAddress1.Add("city", "Seattle");
            msftAddress1.Add("street", "111 South Jackson");

            var msftWebsite1 = new DictionaryTableEntity();
            msftWebsite1.PartitionKey = "MSFT";
            msftWebsite1.RowKey = "WEBSITE-" + Guid.NewGuid().ToString("N").ToUpper();
            msftWebsite1.Add("url", "http://www.microsoft.com");

            var msftWebsite2 = new DictionaryTableEntity();
            msftWebsite2.PartitionKey = "MSFT";
            msftWebsite2.RowKey = "WEBSITE-" + Guid.NewGuid().ToString("N").ToUpper();
            msftWebsite2.Add("url", "http://www.windowsazure.com");

            var batch = new TableBatchOperation();
            batch.Add(TableOperation.Insert(msftAddress1));
            batch.Add(TableOperation.Insert(msftWebsite1));
            batch.Add(TableOperation.Insert(msftWebsite2));
            customersMetadataTable.ExecuteBatch(batch);

            Console.WriteLine("Done. Press ENTER to read the customer metadata.");
            Console.ReadLine();
        }

      private TableBatchOperation ComposeBatch(IEnumerable<LogEvent> events)
      {
         var batch = new TableBatchOperation();

         foreach (LogEvent e in events)
         {
            var row = new ElasticTableEntity
            {
               PartitionKey = e.EventTime.ToString("yy-MM-dd"),
               RowKey = e.EventTime.ToString("HH-mm-ss-fff")
            };

            row.Add("source", e.SourceName);
            row.Add("severity", e.Severity);
            row.Add("message", e.FormattedMessage);
            row.Add("error", e.ErrorException == null ? string.Empty : e.ErrorException.ToString());

            if (e.Properties != null)
            {
               foreach (var p in e.Properties)
               {
                  if (p.Key == LogEvent.ErrorPropertyName) continue;

                  row.Add(p.Key, p.Value);
               }
            }

            batch.Insert(row);
         }

         return batch.Count > 0 ? batch : null;
      }

        protected override void Append(LoggingEvent[] loggingEvents)
        {
            try
            {

                TableBatchOperation batchOperation = new TableBatchOperation();
                base.Append(loggingEvents);
                foreach (var e in loggingEvents)
                {
                    var logitem = new LogItem
                        {
                            Exception = e.GetExceptionString(),
                            Level = e.Level.Name,
                            LoggerName = e.LoggerName,
                            Message = e.RenderedMessage,
                            RoleInstance = RoleInfo
                        };
                    batchOperation.Insert(logitem);
                }
                _log4NetTable.ExecuteBatch(batchOperation);

            }
            catch (Exception ex)
            {
                Trace.TraceError("AzureTableAppender Append " + ex.Message);
            }
        }

 public async Task AddVersion(OwnershipRegistration registration, OwnershipOwner owner, string version)
 {
     CloudTableClient client = _account.CreateCloudTableClient();
     CloudTable table = client.GetTableReference(OwnershipTableName);
     TableBatchOperation batch = new TableBatchOperation();
     batch.InsertOrReplace(new TypedEntity(registration.GetKey(), owner.GetKey(), OwnerType));
     batch.InsertOrReplace(new TypedEntity(registration.GetKey(), version, PackageType));
     await table.ExecuteBatchAsync(batch);
 }

 public void AddEntryToTable(string tableName)
 {
     CloudTable table = cloudTableClient.GetTableReference(tableName);
     table.CreateIfNotExists();
     TableBatchOperation batch = new TableBatchOperation();
     batch.Add(TableOperation.Insert(new TableEntity { RowKey = "Abhishek", PartitionKey = "Kolkata" }));
     batch.Add(TableOperation.Insert(new TableEntity { RowKey = "Abhijit", PartitionKey = "Kolkata" }));
     table.ExecuteBatch(batch);
 }