Use Case

DMF Import and export fails during parallel execution of the Job

Data Management framework works great when a large amount of data should be imported or exported from D365FO. The DMF REST API to import and export doesn’t work when third party applications queue Import and export Jobs in parallel. It results in an unexpected exception from DMF endpoints and failures in the DMF Data Job. The parallel import/export data jobs work for different entities and fail only when the job is for the same entity. E.g. We can import Vendors, Customers and General Journal in Parallel, but we cannot have parallel import of multiple jobs for same the entity.

The following exceptions were discovered during the import

• XML is not in correct format and thus 0 General Journal records are inserted in staging.
• Cannot edit a record in Entities for a processing group (DMFDefinitionGroupEntity).\nThe record has never been selected.
• Cannot delete a record in Source (DMFDefinitionGroupEntityXMLFields).\nDeadlock, where one or more users have simultaneously locked the whole table or part of it.”,
• Exception occurred while executing action ImportFromPackage on Entity DataManagementDefinitionGroup: BOX API can’t be used from non-interactive sessions
• ‘The record already exists’
• “Cannot create a record in Source (DMFDefinitionGroupEntityXMLFields). Entity: General Journal, ACCOUNTINGDATE.\nDeadlock, where one or more users have simultaneously locked the whole table or part of it.”,​

Resolution

Resolution to the problem is to use Recurring Integration. The recurring integration provides

Queuing mechanism for Data Jobs (import/export)

The module will ensure the sequential execution of the Job.

The module provides opportunity to ordered execution of the Job

Recurring Integration

Recurring integration does the following things:

• It builds on data entities and the Data management framework.
• It enables the exchange of documents or files between Finance and Operations and any third-party application or service.
• It supports several document formats, source mapping, Extensible Stylesheet Language Transformations (XSLT), and filters.
• Document/file exchange in several document formats
• It uses secure REST application programming interfaces (APIs) and authorization mechanisms to receive data from, and send data back to, integration systems.

The complete flow to import job to recurring integration is shown below

The set up involves following set and the API supports import/export of DMF Data projects

Authorization for the integration API

The integration REST API uses the same OAuth 2.0 authentication model as the other service endpoints. Before the integrating client application can consume this endpoint, you must create an application ID in Microsoft Azure Active Directory (Azure AD) and give it appropriate permission to the application. When you create and enable a recurring job, you’re prompted to enter the Azure AD application ID that will interact with that recurring job. Therefore, be sure to make a note of the application ID.

 internal static class AuthManager
    {      
        static string aadTenant =  "https://login.windows.net/<<TenantName>>";
        internal static string aadResource =  "https://XXXXX.cloudax.dynamics.com";
        static string aadClientAppId = "The client ID";
        static string aadClientAppSecret = "The Client Secret";

        /// <summary>
        /// Retrieves an authentication header from the service.
        /// </summary>
        /// <returns>The authentication header for the Web API call.</returns>
        internal static string GetAuthenticationHeader()
        {
            AuthenticationContext authenticationContext = new AuthenticationContext(aadTenant);
            var creadential = new ClientCredential(aadClientAppId, aadClientAppSecret);
            AuthenticationResult authenticationResult = authenticationContext.AcquireTokenAsync(aadResource, creadential).Result;
            return authenticationResult.AccessToken;
        }
    }

Set up a data project and recurring data jobs

Create a data project

1. On the main dashboard, select the Data management tile to open the Data management workspace.
2. Select the Import or Export tile to create a new data project.
3. Enter a valid job name, data source, and entity name.
4. Upload a data file for one or more entities. Make sure that each entity is added, and that no errors occur.
5. Select Save.

Create a recurring data job

1. On the Data project page, select Create recurring data job.
2. Enter a valid name and a description for the recurring data job.
3. On the Set-up authorization policy tab, enter the application ID that was generated for your application, and mark it as enabled.
4. Expand Advanced options tab and specify either File or Data package.
5. Select Set processing recurrence, and then, in the Define recurrence dialog box, set up a valid recurrence for your data job
6. Select OK, and then select Yes in the confirmation message box.

Submitting data to recurring data jobs

You can use integration REST endpoints to integrate with the client, submit documents (import), or poll available documents for download (export). These endpoints support OAuth.

Queue the Import Job

API for import (enqueue)

Make an HTTP POST call against the following URL and In the message body, you can the pass the data as a memory stream.

https://<base URL>/api/connector/enqueue/<activity ID>?entity=<entity name>

The following code shows the way to queue the import Job to recurring integration. This approach uses data package-based import. Recurring integration supports both Data package import and the file import. The following parameters will be used

• The D365UO environment
• The Legal entity Name
• The ID of the recurring Job which was created in the previous step
• The entity name which needs to be imported
• Name or description for the import Job

  public static class RecurringIntegration
    {
        /// <summary>
        /// Post request
        /// </summary>
        /// <param name="uri">Enqueue endpoint URI</param>
        /// <param name="authenticationHeader">Authentication header</param>
        /// <param name="bodyStream">Body stream</param>        
        /// <param name="message">ActivityMessage context</param>
        /// <returns></returns>
        public static async Task<HttpResponseMessage> SendPostRequestAsync(Uri uri, string authenticationHeader, Stream bodyStream, string externalCorrelationHeaderValue = null)
        {
            string externalidentifier = "x-ms-dyn-externalidentifier";
            ServicePointManager.SecurityProtocol = SecurityProtocolType.Tls |
                    SecurityProtocolType.Tls11 |
                    SecurityProtocolType.Tls12;

            using (HttpClientHandler handler = new HttpClientHandler() { UseCookies = false })
            {
                using (HttpClient httpClient = new HttpClient(handler))
                {
                    httpClient.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", authenticationHeader);

                    // Add external correlation id header id specified and valid
                    if (!string.IsNullOrEmpty(externalCorrelationHeaderValue))
                    {
                        httpClient.DefaultRequestHeaders.Add(externalidentifier, externalCorrelationHeaderValue);
                    }

                    if (bodyStream != null)
                    {
                        using (StreamContent content = new StreamContent(bodyStream))
                        {
                            return await httpClient.PostAsync(uri, content);
                        }
                    }
                }
            }

            return new HttpResponseMessage()
            {
                Content = new StringContent("Request failed at client.", Encoding.ASCII),
                StatusCode = System.Net.HttpStatusCode.PreconditionFailed
            };
        }

        /// <summary>
        /// Get the Enqueue URI
        /// </summary>
        /// <returns>Enqueue URI</returns>
        private static Uri GetEnqueueUri(string recurringJobId, string legalEntity, string entityName)
        {
            string enviornmentUrl =  "https://XXXXXXX.cloudax.dynamics.com";
            string enqueueUrl = "/api/connector/enqueue/";           
            //access the Connector API
            UriBuilder enqueueUri = new UriBuilder(enviornmentUrl);
            enqueueUri.Path = enqueueUrl + recurringJobId;
            // Data package        
            string enqueueQuery = "entity=" + entityName;
            if (!string.IsNullOrEmpty(legalEntity))
            {
                enqueueQuery += "&company=" + legalEntity;
            }
            enqueueUri.Query = enqueueQuery;        

            return enqueueUri.Uri;
        }
        public static Stream Read(string fullFilePath)
        {
            if (File.Exists(fullFilePath))
            {
                return new FileStream(fullFilePath,
                            FileMode.Open,
                            FileAccess.Read,
                            FileShare.Read,
                            0x1000,
                            true);
            }
            return null;
        }

        /// <summary>
        // Enqueue the Data package to Recurring integration
        /// </summary>
        /// <returns>Status</returns>
        internal static async void QueueImport()
        {
            Stream stream = Read(@"C:\Temp\GL\General Journal.zip");
            string authHeader = AuthManager.GetAuthenticationHeader();
            Uri enqueueUri =GetEnqueueUri("<<ID of the recurring Job>>", "<<Legal Entity>>", "<<Entity Name>>");
            string jobName = "The name of the Job";
            HttpResponseMessage result = SendPostRequestAsync(enqueueUri, authHeader, stream, jobName).Result;
            string resultContent = await result.Content.ReadAsStringAsync();
            Console.WriteLine("Response is");
            Console.WriteLine(resultContent);
        }
    }