diff --git a/src/Sources/CdmChangeFeedSource/Extensions/CsvOperations.cs b/src/Sources/Extensions/CsvOperations.cs
similarity index 93%
rename from src/Sources/CdmChangeFeedSource/Extensions/CsvOperations.cs
rename to src/Sources/Extensions/CsvOperations.cs
index 21bc174..db52b29 100644
--- a/src/Sources/CdmChangeFeedSource/Extensions/CsvOperations.cs
+++ b/src/Sources/Extensions/CsvOperations.cs
@@ -1,8 +1,8 @@
-using System;
+using System;
 using System.Linq;
 using System.Text.RegularExpressions;
 
-namespace Arcane.Framework.Sources.CdmChangeFeedSource.Extensions;
+namespace Arcane.Framework.Sources.Extensions;
 
 /// <summary>
 /// Contains operations for parsing CSV files.
@@ -16,7 +16,7 @@ internal static class CsvOperations
     /// <param name="headerCount">Number of expected headers.</param>
     /// <param name="delimiter">Delimiter used in the line.</param>
     /// <returns>A string array of individual values.</returns>
-    public static string[] ParseCsvLine(string line, int headerCount, char delimiter = ',')
+    public static string[] ParseCsvLine(this string line, int headerCount, char delimiter = ',')
     {
         var result = new string[headerCount];
         var fieldCounter = 0;
@@ -91,7 +91,7 @@ public static bool IsComplete(string csvLine)
     /// </summary>
     /// <param name="csvLine">A CSV line to purge newlines from.</param>
     /// <returns>A CSV line without newline characters inside column values.</returns>
-    public static string ReplaceQuotedNewlines(string csvLine)
+    public static string ReplaceQuotedNewlines(this string csvLine)
     {
         return Regex.Replace(csvLine, "\"[^\"]*(?:\"\"[^\"]*)*\"", m => m.Value.Replace("\n", "")).Replace("\r", "");
     }
diff --git a/src/Sources/SalesForce/Exceptions/SalesForceJobAbortedException.cs b/src/Sources/SalesForce/Exceptions/SalesForceJobAbortedException.cs
new file mode 100644
index 0000000..d2838e7
--- /dev/null
+++ b/src/Sources/SalesForce/Exceptions/SalesForceJobAbortedException.cs
@@ -0,0 +1,19 @@
+using System;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Arcane.Framework.Sources.SalesForce.Exceptions;
+
+/// <summary>
+/// Thrown if the Salesforce job was aborted.
+/// </summary>
+[ExcludeFromCodeCoverage(Justification = "Trivial")]
+public class SalesForceJobAbortedException : SalesForceJobException
+{
+    public SalesForceJobAbortedException(string message) : base(message)
+    {
+    }
+
+    public SalesForceJobAbortedException(string message, Exception inner) : base(message, inner)
+    {
+    }
+}
diff --git a/src/Sources/SalesForce/Exceptions/SalesForceJobException.cs b/src/Sources/SalesForce/Exceptions/SalesForceJobException.cs
new file mode 100644
index 0000000..2412bdc
--- /dev/null
+++ b/src/Sources/SalesForce/Exceptions/SalesForceJobException.cs
@@ -0,0 +1,21 @@
+using System;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Arcane.Framework.Sources.SalesForce.Exceptions;
+
+/// <summary>
+/// Base class for all Salesforce-related exceptions
+/// </summary>
+[ExcludeFromCodeCoverage(Justification = "Trivial")]
+public class SalesForceJobException : Exception
+{
+    public SalesForceJobException(string message)
+        : base(message)
+    {
+    }
+
+    public SalesForceJobException(string message, Exception inner)
+        : base(message, inner)
+    {
+    }
+}
diff --git a/src/Sources/SalesForce/Exceptions/SalesForceJobFailedException.cs b/src/Sources/SalesForce/Exceptions/SalesForceJobFailedException.cs
new file mode 100644
index 0000000..18b1ffa
--- /dev/null
+++ b/src/Sources/SalesForce/Exceptions/SalesForceJobFailedException.cs
@@ -0,0 +1,19 @@
+using System;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Arcane.Framework.Sources.SalesForce.Exceptions;
+
+/// <summary>
+/// Thrown if the Salesforce job return with failed status.
+/// </summary>
+[ExcludeFromCodeCoverage(Justification = "Trivial")]
+public class SalesForceJobFailedException : SalesForceJobException
+{
+    public SalesForceJobFailedException(string message) : base(message)
+    {
+    }
+
+    public SalesForceJobFailedException(string message, Exception inner) : base(message, inner)
+    {
+    }
+}
diff --git a/src/Sources/SalesForce/Models/SalesForceAttribute.cs b/src/Sources/SalesForce/Models/SalesForceAttribute.cs
new file mode 100644
index 0000000..74050b3
--- /dev/null
+++ b/src/Sources/SalesForce/Models/SalesForceAttribute.cs
@@ -0,0 +1,93 @@
+using System;
+using System.Collections.Generic;
+using System.Text.Json.Serialization;
+
+namespace Arcane.Framework.Sources.SalesForce.Models;
+
+/// <summary>
+/// Represents Salesforce attribute
+/// </summary>
+public class SalesForceAttribute
+{
+    private static readonly Dictionary<string, Type> salesforceTypeMap = new()
+    {
+        { "string", typeof(string) },
+        { "id", typeof(string) },
+        { "datetime", typeof(DateTime) },
+        { "date", typeof(DateTime) },
+        { "decimal", typeof(decimal) },
+        { "integer", typeof(int) },
+        { "long", typeof(long) },
+        { "double", typeof(double) },
+        { "boolean", typeof(bool) },
+    };
+
+    /// <summary>
+    /// Attribute name
+    /// </summary>
+    [JsonPropertyName("Name")]
+    public string Name { get; set; }
+
+    /// <summary>
+    /// Attribute data type
+    /// </summary>
+    [JsonPropertyName("ValueTypeId")]
+    public string DataType { get; set; }
+
+
+    /// <summary>
+    /// Attribute comparer
+    /// </summary>
+    public static IEqualityComparer<SalesForceAttribute> SalesForceAttributeComparer { get; } =
+        new SalesForceAttributeEqualityComparer();
+
+    /// <summary>
+    /// Maps Salesforce type to .NET type
+    /// </summary>
+    /// <param name="salesforceTypeName">Salesforce type name</param>
+    /// <returns>.NET type instance</returns>
+    /// <exception cref="InvalidOperationException">Thrown if type is not supported</exception>
+    public static Type MapSalesforceType(string salesforceTypeName)
+    {
+        if (salesforceTypeMap.ContainsKey(salesforceTypeName.ToLower()))
+        {
+            return salesforceTypeMap[salesforceTypeName.ToLower()];
+        }
+
+        throw new InvalidOperationException($"Unsupported type: {salesforceTypeName}");
+    }
+
+    private sealed class SalesForceAttributeEqualityComparer : IEqualityComparer<SalesForceAttribute>
+    {
+        public bool Equals(SalesForceAttribute x, SalesForceAttribute y)
+        {
+            if (ReferenceEquals(x, y))
+            {
+                return true;
+            }
+
+            if (ReferenceEquals(x, null))
+            {
+                return false;
+            }
+
+            if (ReferenceEquals(y, null))
+            {
+                return false;
+            }
+
+            if (x.GetType() != y.GetType())
+            {
+                return false;
+            }
+
+            return x.Name == y.Name &&
+                   x.DataType == y.DataType;
+        }
+
+        public int GetHashCode(SalesForceAttribute obj)
+        {
+            return HashCode.Combine(obj.Name, obj.DataType);
+        }
+    }
+}
diff --git a/src/Sources/SalesForce/Models/SalesForceEntity.cs b/src/Sources/SalesForce/Models/SalesForceEntity.cs
new file mode 100644
index 0000000..7977f3a
--- /dev/null
+++ b/src/Sources/SalesForce/Models/SalesForceEntity.cs
@@ -0,0 +1,98 @@
+using System;
+using System.Collections.Generic;
+using System.Data;
+using System.Linq;
+using System.Text.Json;
+
+namespace Arcane.Framework.Sources.SalesForce.Models;
+
+/// <summary>
+/// Represents Salesforce entity
+/// </summary>
+public class SalesForceEntity
+{
+    /// <summary>
+    /// Entity name
+    /// </summary>
+    public string EntityName { get; set; }
+
+
+    /// <summary>
+    /// Attributes collection
+    /// </summary>
+    public SalesForceAttribute[] Attributes { get; set; }
+
+    /// <summary>
+    /// Comparer class
+    /// </summary>
+    public static IEqualityComparer<SalesForceEntity> SalesForceEntityComparer { get; } =
+        new SalesForceEntityEqualityComparer();
+
+    /// <summary>
+    /// Parse Salesforce entity from a JSON document
+    /// </summary>
+    /// <param name="entityName">Name of the Salesforce entity </param>
+    /// <param name="document">Json document to parse</param>
+    /// <returns>Parsed SalesForceEntity object</returns>
+    public static SalesForceEntity FromJson(string entityName, JsonDocument document)
+    {
+        var entity = new SalesForceEntity
+        {
+            EntityName = entityName,
+            Attributes = document.RootElement.GetProperty("records").Deserialize<SalesForceAttribute[]>()
+        };
+
+
+        return entity;
+    }
+
+    /// <summary>
+    /// Create DataReader for the entity
+    /// </summary>
+    /// <returns>DataReader instance</returns>
+    public IDataReader GetReader()
+    {
+        var dt = new DataTable();
+
+        foreach (var attr in this.Attributes)
+        {
+            dt.Columns.Add(new DataColumn(attr.Name, SalesForceAttribute.MapSalesforceType(attr.DataType)));
+        }
+
+        return dt.CreateDataReader();
+    }
+
+    private sealed class SalesForceEntityEqualityComparer : IEqualityComparer<SalesForceEntity>
+    {
+        public bool Equals(SalesForceEntity x, SalesForceEntity y)
+        {
+            if (ReferenceEquals(x, y))
+            {
+                return true;
+            }
+
+            if (ReferenceEquals(x, null))
+            {
+                return false;
+            }
+
+            if (ReferenceEquals(y, null))
+            {
+                return false;
+            }
+
+            if (x.GetType() != y.GetType())
+            {
+                return false;
+            }
+
+            return x.EntityName == y.EntityName
+                   && x.Attributes.SequenceEqual(y.Attributes, SalesForceAttribute.SalesForceAttributeComparer);
+        }
+
+        public int GetHashCode(SalesForceEntity obj)
+        {
+            return HashCode.Combine(obj.EntityName, obj.Attributes);
+        }
+    }
+}
diff --git a/src/Sources/SalesForce/Models/SalesForceJob.cs b/src/Sources/SalesForce/Models/SalesForceJob.cs
new file mode 100644
index 0000000..8d4ad26
--- /dev/null
+++ b/src/Sources/SalesForce/Models/SalesForceJob.cs
@@ -0,0 +1,55 @@
+using System.Text.Json.Serialization;
+
+namespace Arcane.Framework.Sources.SalesForce.Models;
+
+/// <summary>
+/// Job status types
+/// </summary>
+[JsonConverter(typeof(JsonStringEnumConverter))]
+public enum SalesforceJobStatus
+{
+    UploadComplete,
+    InProgress,
+    Aborted,
+    JobComplete,
+    Failed,
+    None
+}
+
+/// <summary>
+/// Represents Salesforce job
+/// </summary>
+public class SalesForceJob
+{
+    /// <summary>
+    /// Id
+    /// </summary>
+    [JsonPropertyName("id")]
+    public string Id { get; set; }
+
+
+    /// <summary>
+    /// Job status
+    /// </summary>
+    [JsonPropertyName("state")]
+    public SalesforceJobStatus Status { get; set; }
+
+    /// <summary>
+    /// object
+    /// </summary>
+    [JsonPropertyName("object")]
+    public string Object { get; set; }
+
+    /// <summary>
+    /// Total processing time of the job
+    /// </summary>
+    [JsonPropertyName("totalProcessingTime")]
+    public long? TotalProcessingTime { get; set; }
+
+    /// <summary>
+    /// Numbers of records processed by the job
+    /// </summary>
+    [JsonPropertyName("numberRecordsProcessed")]
+    public long? NumberRecordsProcessed { get; set; }
+
+}
diff --git a/src/Sources/SalesForce/SalesForceSource.cs b/src/Sources/SalesForce/SalesForceSource.cs
new file mode 100644
index 0000000..25038a9
--- /dev/null
+++ b/src/Sources/SalesForce/SalesForceSource.cs
@@ -0,0 +1,264 @@
+using System;
+using System.Collections.Generic;
+using System.Configuration;
+using System.IO;
+using System.Linq;
+using System.Net.Http;
+using Akka.Actor;
+using Akka.Event;
+using Akka.Streams;
+using Akka.Streams.Stage;
+using Akka.Util;
+using Arcane.Framework.Contracts;
+using Arcane.Framework.Sinks.Parquet;
+using Arcane.Framework.Sinks.Parquet.Models;
+using Arcane.Framework.Sources.Base;
+using Arcane.Framework.Sources.Exceptions;
+using Arcane.Framework.Sources.SalesForce.Models;
+using Arcane.Framework.Sources.SalesForce.Services.AuthenticatedMessageProviders;
+using Arcane.Framework.Sources.SalesForce.Exceptions;
+using Parquet.Data;
+
+namespace Arcane.Framework.Sources.SalesForce;
+
+/// <summary>
+/// Source for reading data from SalesForce BULK v2 API.
+/// </summary>
+public class SalesForceSource : GraphStage<SourceShape<List<DataCell>>>, IParquetSource, ITaggedSource
+{
+    private readonly string entityName;
+    private readonly SalesForceJobProvider jobProvider;
+    private readonly HttpClient httpClient;
+
+    private readonly TimeSpan changeCaptureInterval;
+
+    private SalesForceSource(
+        SalesForceJobProvider jobProvider,
+        string entityName,
+        TimeSpan changeCaptureInterval
+        )
+    {
+        this.jobProvider = jobProvider;
+        this.httpClient = new HttpClient();
+        this.entityName = entityName;
+        this.changeCaptureInterval = changeCaptureInterval;
+
+        this.Shape = new SourceShape<List<DataCell>>(this.Out);
+    }
+
+
+    /// <summary>
+    /// Only use this constructor for unit tests to mock http calls.
+    /// </summary>
+    /// <param name="jobProvider">Salesforce Job Provider</param>
+    /// <param name="entityName">Name of Salesforce entity</param>
+    /// <param name="httpClient">Http client for making requests</param>
+    /// <param name="changeCaptureInterval">How often to track changes</param>
+    /// <returns></returns>
+    private SalesForceSource(
+
+        SalesForceJobProvider jobProvider,
+        string entityName,
+        HttpClient httpClient,
+        TimeSpan changeCaptureInterval
+        ) : this(jobProvider, entityName, changeCaptureInterval)
+    {
+        this.httpClient = httpClient;
+    }
+
+
+    /// <inheritdoc cref="GraphStageWithMaterializedValue{TShape,TMaterialized}.InitialAttributes"/>
+    protected override Attributes InitialAttributes { get; } = Attributes.CreateName(nameof(SalesForceSource));
+
+    /// <summary>
+    /// Source outlet
+    /// </summary>
+    public Outlet<List<DataCell>> Out { get; } = new($"{nameof(SalesForceSource)}.Out");
+
+    /// <inheritdoc cref="GraphStageWithMaterializedValue{TShape,TMaterialized}.Shape"/>
+    public override SourceShape<List<DataCell>> Shape { get; }
+
+    /// <inheritdoc cref="IParquetSource.GetParquetSchema"/>
+    public Schema GetParquetSchema()
+    {
+        var schema = this.jobProvider.GetSchema(httpClient, this.entityName).Result;
+
+        return schema.Value.GetReader().ToParquetSchema();
+
+    }
+
+    /// <inheritdoc cref="ITaggedSource.GetDefaultTags"/>
+    public SourceTags GetDefaultTags()
+    {
+        return new SourceTags
+        {
+            SourceEntity = this.entityName,
+            SourceLocation = this.jobProvider.accountName
+
+        };
+    }
+
+    /// <summary>
+    /// Creates a new instance of <see cref="SalesForceSource"/>
+    /// </summary>
+    /// <param name="jobProvider"></param>
+    /// <param name="httpClient"></param>
+    /// <param name="entityName"></param>
+    /// <param name="changeCaptureInterval"></param>
+    public static SalesForceSource Create(
+        SalesForceJobProvider jobProvider,
+        HttpClient httpClient,
+        string entityName,
+        TimeSpan changeCaptureInterval
+     )
+    {
+        return new SalesForceSource(jobProvider
+            , entityName, httpClient, changeCaptureInterval);
+    }
+
+
+
+    /// <inheritdoc cref="GraphStage{TShape}.CreateLogic"/>
+    protected override GraphStageLogic CreateLogic(Attributes inheritedAttributes)
+    {
+        return new SourceLogic(this);
+    }
+
+    private sealed class SourceLogic : TimerGraphStageLogic
+    {
+        private const string TimerKey = nameof(SalesForceSource);
+        private readonly LocalOnlyDecider decider;
+        private readonly HttpClient httpClient;
+        private readonly SalesForceSource source;
+        private SalesForceEntity entitySchema;
+        private Option<SalesForceJob> currentJob;
+        private Option<string> nextJobLocator;
+
+
+
+        public SourceLogic(SalesForceSource source) : base(source.Shape)
+        {
+            this.source = source;
+
+            this.httpClient = this.source.httpClient ?? new HttpClient();
+
+            this.decider = Decider.From((ex) => ex.GetType().Name switch
+            {
+                nameof(ArgumentException) => Directive.Stop,
+                nameof(ArgumentNullException) => Directive.Stop,
+                nameof(InvalidOperationException) => Directive.Stop,
+                nameof(ConfigurationErrorsException) => Directive.Stop,
+                nameof(ObjectDisposedException) => Directive.Stop,
+                nameof(IOException) => Directive.Restart,
+                nameof(TimeoutException) => Directive.Restart,
+                nameof(HttpRequestException) => Directive.Restart,
+                _ => Directive.Stop
+            });
+
+            this.currentJob = Option<SalesForceJob>.None;
+            this.nextJobLocator = Option<string>.None;
+
+
+            this.SetHandler(source.Out, this.PullChanges, this.Finish);
+        }
+
+        private void Finish(Exception cause)
+        {
+            if (cause is not null && cause is not SubscriptionWithCancelException.NonFailureCancellation)
+            {
+                this.FailStage(cause);
+            }
+
+            this.httpClient.Dispose();
+        }
+
+        public override void PreStart()
+        {
+            this.UpdateSchema();
+        }
+
+        public void UpdateSchema()
+        {
+            var newSchema = this.source.jobProvider.GetSchema(httpClient, this.source.entityName).Result;
+
+            if (newSchema.IsEmpty)
+            {
+                this.Log.Warning("Could not update schema");
+            }
+            else
+            {
+                var schemaUnchanged =
+                    SalesForceEntity.SalesForceEntityComparer.Equals(this.entitySchema, newSchema.Value);
+                this.entitySchema = (this.entitySchema == null, schemaEquals: schemaUnchanged) switch
+                {
+                    (true, _) or (false, true) => newSchema.Value,
+                    (false, false) => throw new SchemaMismatchException()
+                };
+            }
+
+        }
+
+        private void CreateNewJob()
+        {
+            var job = this.source.jobProvider.CreateJob(httpClient, this.entitySchema).Result;
+
+            if (job.IsEmpty)
+            {
+                this.Log.Warning("Could not create job");
+            }
+            else
+            {
+                this.currentJob = job;
+            }
+            this.ScheduleOnce(TimerKey, TimeSpan.FromSeconds(1));
+        }
+
+        private void UpdateJobStatus()
+        {
+            var response = this.source.jobProvider.GetJobStatus(httpClient, this.currentJob.Value).Result;
+
+            if (response.IsEmpty)
+            {
+                this.Log.Warning("Could not create job");
+            }
+            else
+            {
+                this.currentJob = response;
+            }
+            this.ScheduleOnce(TimerKey, TimeSpan.FromSeconds(1));
+        }
+
+        private void ProcessResult()
+        {
+
+            var (rows, nextJobLocator) = this.source.jobProvider.GetJobResult(httpClient, this.currentJob.Value, this.entitySchema, this.nextJobLocator).Result;
+            this.nextJobLocator = nextJobLocator;
+            this.EmitMultiple(this.source.Out, rows);
+
+            if (nextJobLocator.IsEmpty)
+            {
+                this.currentJob = Option<SalesForceJob>.None;
+                this.ScheduleOnce(TimerKey, this.source.changeCaptureInterval);
+            }
+
+        }
+
+        private void PullChanges()
+        {
+            switch (this.currentJob.Select(job => job.Status).GetOrElse(SalesforceJobStatus.None))
+            {
+                case SalesforceJobStatus.UploadComplete:; this.UpdateJobStatus(); break;
+                case SalesforceJobStatus.InProgress: this.UpdateJobStatus(); break;
+                case SalesforceJobStatus.Aborted: this.FailStage(new SalesForceJobAbortedException($"job : {this.currentJob.Value.Id} was aborted by source")); break;
+                case SalesforceJobStatus.Failed: this.FailStage(new SalesForceJobFailedException($"job : {this.currentJob.Value.Id} returned with failure")); break;
+                case SalesforceJobStatus.JobComplete: this.ProcessResult(); break;
+                case SalesforceJobStatus.None: this.CreateNewJob(); break;
+            };
+        }
+
+        protected override void OnTimer(object timerKey)
+        {
+            this.PullChanges();
+        }
+    }
+}
diff --git a/src/Sources/SalesForce/Services/AuthenticatedMessageProviders/SalesForceJobProvider.cs b/src/Sources/SalesForce/Services/AuthenticatedMessageProviders/SalesForceJobProvider.cs
new file mode 100644
index 0000000..9d0dfde
--- /dev/null
+++ b/src/Sources/SalesForce/Services/AuthenticatedMessageProviders/SalesForceJobProvider.cs
@@ -0,0 +1,316 @@
+using System;
+using System.Collections.Generic;
+using System.ComponentModel;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Net.Http.Json;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.Util;
+using Akka.Util.Extensions;
+using Arcane.Framework.Sinks.Parquet.Models;
+using Arcane.Framework.Sources.Extensions;
+using Arcane.Framework.Sources.SalesForce.Models;
+using Snd.Sdk.Tasks;
+
+namespace Arcane.Framework.Sources.SalesForce.Services.AuthenticatedMessageProviders;
+
+/// <summary>
+/// Salesforce Job provider that handles talking with the Bulk V2 API
+/// </summary>
+public record SalesForceJobProvider
+{
+    private readonly TimeSpan expirationPeriod;
+    private string currentToken;
+    private DateTimeOffset? validTo;
+    private readonly Uri tokenSource;
+    /// <summary>
+    /// Salesforce Account
+    /// </summary>
+    public readonly string accountName;
+    private readonly string clientId;
+    private readonly string clientSecret;
+    private readonly string username;
+    private readonly string password;
+    private readonly string securityToken;
+    private readonly string apiVersion;
+    private readonly Option<int> rowsPerPage;
+
+    /// <summary>
+    /// Salesforce Job provider that handles talking with the Bulk V2 API.
+    /// </summary>
+
+    public SalesForceJobProvider(string accountName, string clientId, string clientSecret, string username, string password, string securityToken, string apiVersion, int rowsPerPage)
+    {
+        this.tokenSource = new Uri($"https://{accountName}/services/oauth2/token");
+        this.accountName = accountName;
+        this.clientId = clientId;
+        this.clientSecret = clientSecret;
+        this.username = username;
+        this.password = password;
+        this.securityToken = securityToken;
+        this.apiVersion = apiVersion;
+        this.rowsPerPage = rowsPerPage;
+    }
+
+    /// <summary>
+    /// Salesforce Job provider that handles talking with the Bulk V2 API
+    /// </summary>
+    /// <param name="accountName"></param>
+    /// <param name="clientId"></param>
+    /// <param name="clientSecret"></param>
+    /// <param name="username"></param>
+    /// <param name="password"></param>
+    /// <param name="securityToken"></param>
+    /// <param name="apiVersion"></param>
+    public SalesForceJobProvider(string accountName, string clientId, string clientSecret, string username, string password, string securityToken, string apiVersion)
+    {
+        this.tokenSource = new Uri($"https://{accountName}/services/oauth2/token");
+        this.accountName = accountName;
+        this.clientId = clientId;
+        this.clientSecret = clientSecret;
+        this.username = username;
+        this.password = password;
+        this.securityToken = securityToken;
+        this.apiVersion = apiVersion;
+        this.rowsPerPage = Option<int>.None;
+        this.expirationPeriod = TimeSpan.FromMinutes(19);
+    }
+
+    /// <summary>
+    /// Generates authenticated message for the REST API request.
+    /// </summary>
+    /// <param name="httpClient">HTTP client</param>
+    /// <returns>Authenticated message</returns>
+    public Task<HttpRequestMessage> GetAuthenticatedMessage(HttpClient httpClient)
+    {
+        if (this.validTo.GetValueOrDefault(DateTimeOffset.MaxValue) <
+            DateTimeOffset.UtcNow.Subtract(TimeSpan.FromMinutes(1)))
+        {
+            return Task.FromResult(new HttpRequestMessage
+            {
+                Headers = { Authorization = new AuthenticationHeaderValue("Bearer", this.currentToken) }
+            });
+        }
+        var dict = new Dictionary<string, string>
+        {
+            { "grant_type", "password" },
+            { "client_id", this.clientId },
+            { "client_secret", this.clientSecret },
+            { "username", this.username },
+            { "password", $"{this.password}{this.securityToken}" },
+        };
+        var tokenHrm = new HttpRequestMessage(HttpMethod.Post, this.tokenSource)
+        {
+            Content = new FormUrlEncodedContent(dict)
+        };
+
+        return httpClient.SendAsync(tokenHrm, CancellationToken.None).Map(response =>
+        {
+            response.EnsureSuccessStatusCode();
+            return response.Content.ReadAsStringAsync();
+        }).FlatMap(result =>
+        {
+            var tokenResponse = JsonSerializer.Deserialize<JsonElement>(result);
+            this.currentToken = tokenResponse.GetProperty("access_token").GetString();
+            this.validTo = DateTimeOffset.UtcNow.Add(this.expirationPeriod);
+            return new HttpRequestMessage
+            {
+                Headers = { Authorization = new AuthenticationHeaderValue("Bearer", this.currentToken) }
+            };
+        });
+    }
+
+    /// <summary>
+    /// Get the schema of a Salesforce entity
+    /// </summary>
+    /// <param name="httpClient"></param>
+    /// <param name="entityName"></param>
+    public Task<Option<SalesForceEntity>> GetSchema(HttpClient httpClient, string entityName)
+    {
+        return this.GetAuthenticatedMessage(httpClient).Map(msg =>
+
+            {
+                msg.RequestUri = new Uri($"https://{this.accountName}/services/data/{this.apiVersion}/query?q=SELECT Name,DataType,ValueTypeId FROM EntityParticle WHERE EntityDefinition.QualifiedApiName ='{entityName}' and dataType != 'address'");
+
+                return httpClient.SendAsync(msg, default(CancellationToken)).Map(response =>
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        return response.Content.ReadAsStringAsync().Map(value =>
+                        {
+
+                            return SalesForceEntity.FromJson(entityName, JsonSerializer.Deserialize<JsonDocument>(value)).AsOption();
+
+                        });
+                    }
+
+                    var errorMsg = $"API request to {msg.RequestUri} failed with {response.StatusCode}, reason: {response.ReasonPhrase}, content: {response.Content.ReadAsStringAsync().ConfigureAwait(false).GetAwaiter().GetResult()}";
+
+
+                    throw new HttpRequestException(errorMsg, null, response.StatusCode);
+                }).Flatten();
+            }).Flatten().TryMap(result => result, exception => exception switch
+            {
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.TooManyRequests
+                } => Option<SalesForceEntity>.None, // API rate limit, in case configured rate limit is not good enough
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.RequestTimeout
+                } => Option<SalesForceEntity>.None, // Potential server-side timeout due to overload
+                _ => throw exception
+            });
+    }
+
+    /// <summary>
+    /// Submit a new query job
+    /// </summary>
+    /// <param name="httpClient"></param>
+    /// <param name="entitySchema"></param>
+    public Task<Option<SalesForceJob>> CreateJob(HttpClient httpClient, SalesForceEntity entitySchema)
+    {
+        return this.GetAuthenticatedMessage(httpClient).Map(msg =>
+
+            {
+                msg.RequestUri = new Uri($"https://{this.accountName}/services/data/{this.apiVersion}/jobs/query");
+                msg.Content = JsonContent.Create(new
+                {
+                    operation = "query",
+                    query = $"SELECT {entitySchema.Attributes.Where(e => e.DataType != "address").Select(e => e.Name).Aggregate((a, b) => a + ", " + b)} FROM {entitySchema.EntityName}",
+
+                });
+                msg.Method = HttpMethod.Post;
+
+                return httpClient.SendAsync(msg, default(CancellationToken)).Map(response =>
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        return response.Content.ReadAsStringAsync().Map(value =>
+                        {
+
+                            return JsonSerializer.Deserialize<SalesForceJob>(value).AsOption();
+
+                        });
+                    }
+
+                    var errorMsg = $"API request to {msg.RequestUri} failed with {response.StatusCode}, reason: {response.ReasonPhrase}, content: {response.Content.ReadAsStringAsync().ConfigureAwait(false).GetAwaiter().GetResult()}";
+
+                    throw new HttpRequestException(errorMsg, null, response.StatusCode);
+                }).Flatten();
+            }).Flatten().TryMap(result => result, exception => exception switch
+            {
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.TooManyRequests
+                } => Option<SalesForceJob>.None, // API rate limit, in case configured rate limit is not good enough
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.RequestTimeout
+                } => Option<SalesForceJob>.None, // Potential server-side timeout due to overload
+                _ => throw exception
+            });
+    }
+
+    /// <summary>
+    /// Get status of query job
+    /// </summary>
+    /// <param name="httpClient"></param>
+    /// <param name="job"></param>
+    public Task<Option<SalesForceJob>> GetJobStatus(HttpClient httpClient, SalesForceJob job)
+    {
+        return this.GetAuthenticatedMessage(httpClient).Map(msg =>
+
+            {
+                msg.RequestUri = new Uri($"https://{this.accountName}/services/data/{this.apiVersion}/jobs/query/{job.Id}");
+
+                return httpClient.SendAsync(msg, default(CancellationToken)).Map(response =>
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        return response.Content.ReadAsStringAsync().Map(value =>
+                        {
+
+                            return JsonSerializer.Deserialize<SalesForceJob>(value).AsOption();
+
+                        });
+                    }
+
+                    var errorMsg = $"API request to {msg.RequestUri} failed with {response.StatusCode}, reason: {response.ReasonPhrase}, content: {response.Content.ReadAsStringAsync().ConfigureAwait(false).GetAwaiter().GetResult()}";
+
+                    throw new HttpRequestException(errorMsg, null, response.StatusCode);
+                }).Flatten();
+            }).Flatten().TryMap(result => result, exception => exception switch
+            {
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.TooManyRequests
+                } => Option<SalesForceJob>.None, // API rate limit, in case configured rate limit is not good enough
+                HttpRequestException
+                {
+                    StatusCode: HttpStatusCode.RequestTimeout
+                } => Option<SalesForceJob>.None, // Potential server-side timeout due to overload
+                _ => throw exception
+            });
+    }
+
+    private (Type, object) ConvertToSalesforceType(string salesforceDataType, string value)
+    {
+        var tp = SalesForceAttribute.MapSalesforceType(salesforceDataType);
+        var converter = TypeDescriptor.GetConverter(tp);
+        return (tp, value == "" ? null : converter.ConvertFromInvariantString(value));
+    }
+
+    /// <summary>
+    /// Get result of query job
+    /// </summary>
+    /// <param name="httpClient"></param>
+    /// <param name="job"></param>
+    /// <param name="entitySchema"></param>
+    public Task<(IEnumerable<List<DataCell>>, Option<string>)> GetJobResult(HttpClient httpClient, SalesForceJob job, SalesForceEntity entitySchema, Option<string> jobLocator)
+    {
+        return this.GetAuthenticatedMessage(httpClient).Map(msg =>
+
+            {
+                var maxRowString = this.rowsPerPage.HasValue ? $"maxRecords={this.rowsPerPage.Value}" : "";
+                var locatorString = jobLocator.HasValue ? $"locator={jobLocator.Value}" : "";
+                var urlParams = string.Join("&", new[] { maxRowString, locatorString });
+                msg.RequestUri = new Uri($"https://{this.accountName}/services/data/{this.apiVersion}/jobs/query/{job.Id}/results?{urlParams}");
+                msg.Headers.Add("Accept", "text/csv");
+
+                return httpClient.SendAsync(msg, default(CancellationToken)).Map(response =>
+                {
+                    if (response.IsSuccessStatusCode)
+                    {
+                        return response.Content.ReadAsStringAsync().Map(value =>
+                        {
+
+                            var newjobLocator = response.Headers.GetValues("Sforce-Locator").Select(h => h == "null" ? Option<string>.None : h.AsOption()).First();
+                            var rows = value.ReplaceQuotedNewlines().Split("\n").Skip(1).Where(line => !string.IsNullOrEmpty(line)).Select(line =>
+                            {
+                                var cells = line.ParseCsvLine(entitySchema.Attributes.Length).Select(
+                                        (v, ix) =>
+                                        {
+                                            var (tp, value) = this.ConvertToSalesforceType(entitySchema.Attributes[ix].DataType, v);
+                                            return new DataCell(entitySchema.Attributes[ix].Name, tp,
+                                                value);
+                                        }).ToList();
+                                return cells;
+                            });
+                            return (rows, newjobLocator);
+
+
+                        });
+                    }
+
+                    var errorMsg = $"API request to {msg.RequestUri} failed with {response.StatusCode}, reason: {response.ReasonPhrase}, content: {response.Content.ReadAsStringAsync().ConfigureAwait(false).GetAwaiter().GetResult()}";
+
+                    throw new HttpRequestException(errorMsg, null, response.StatusCode);
+                }).Flatten();
+            }).Flatten();
+    }
+}
diff --git a/test/Operations/CsvOperationsTests.cs b/test/Operations/CsvOperationsTests.cs
index cbb4696..de35f7d 100644
--- a/test/Operations/CsvOperationsTests.cs
+++ b/test/Operations/CsvOperationsTests.cs
@@ -1,5 +1,5 @@
 using System;
-using Arcane.Framework.Sources.CdmChangeFeedSource.Extensions;
+using Arcane.Framework.Sources.Extensions;
 using Xunit;
 
 namespace Arcane.Framework.Tests.Operations;
diff --git a/test/Sources/SalesforceSourceTests.cs b/test/Sources/SalesforceSourceTests.cs
new file mode 100644
index 0000000..5cf66b0
--- /dev/null
+++ b/test/Sources/SalesforceSourceTests.cs
@@ -0,0 +1,224 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Akka.Streams.Dsl;
+using Arcane.Framework.Sinks.Parquet.Models;
+using Arcane.Framework.Sources.SalesForce;
+using Arcane.Framework.Sources.SalesForce.Services.AuthenticatedMessageProviders;
+using Arcane.Framework.Tests.Fixtures;
+using Moq;
+using Xunit;
+
+namespace Arcane.Framework.Tests.Sources;
+
+public class SalesforceSourceTests : IClassFixture<AkkaFixture>
+{
+    private readonly AkkaFixture akkaFixture;
+
+    private readonly SalesForceJobProvider jobProvider;
+    private readonly Mock<HttpClient> mockHttp;
+
+
+    public SalesforceSourceTests(AkkaFixture akkaFixture)
+    {
+        this.akkaFixture = akkaFixture;
+        this.mockHttp = new Mock<HttpClient>();
+        this.jobProvider = new SalesForceJobProvider("test.my.salesforce.com", "client_id", "client_secret", "user_name", "password", "security_token", "v60.0", 5);
+
+    }
+
+    [Fact]
+    public async Task TokenGeneration()
+    {
+        var mockContent = new
+        {
+            access_token = "<long_token>",
+            instance_url = "https://test.my.salesforce.com",
+            id = "https://test.salesforce.com/id/00D3E000000D2OCUA0/0057Y0000068CtmQAE",
+            token_type = "Bearer",
+            issued_at = "1718179384536",
+            signature = "abc"
+        };
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/oauth2/token")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK)
+            { Content = new StringContent(JsonSerializer.Serialize(mockContent)) });
+
+
+        var token = await this.jobProvider.GetAuthenticatedMessage(this.mockHttp.Object);
+        Assert.Equal("<long_token>", token.Headers.Authorization.Parameter);
+    }
+
+
+    [Fact]
+    public async void RunStream()
+    {
+
+        var mockTokenContent = new
+        {
+            access_token = "<long_token>",
+            instance_url = "https://test.my.salesforce.com",
+            id = "https://test.salesforce.com/id/00D3E000000D2OCUA0/0057Y0000068CtmQAE",
+            token_type = "Bearer",
+            issued_at = "1718179384536",
+            signature = "abc"
+        };
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/oauth2/token")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK)
+            { Content = new StringContent(JsonSerializer.Serialize(mockTokenContent)) });
+
+        var mockCreateJobContent = new
+        {
+            id = "750QI000007QKdRYAW",
+            operation = "query",
+            @object = "ECCO_Consent__c",
+            createdById = "0057Y0000068CtmQAE",
+            createdDate = "2024-06-13T06:57:56.000+0000",
+            systemModstamp = "2024-06-13T06:57:56.000+0000",
+            state = "UploadComplete",
+            concurrencyMode = "Parallel",
+            contentType = "CSV",
+            apiVersion = 60.0,
+            lineEnding = "LF",
+            columnDelimiter = "COMMA"
+        };
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/data/v60.0/jobs/query") &&
+                        msg.Method == HttpMethod.Post),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK)
+            { Content = new StringContent(JsonSerializer.Serialize(mockCreateJobContent)) });
+
+        var mockGetSchemaContent = new
+        {
+            totalSize = 2,
+            done = true,
+            records = new[]{
+                new {
+
+                    Name = "Id",
+                    @DataType = "id",
+                    ValueTypeId = "id"
+                },
+                new {
+
+                    Name = "Name",
+                    @DataType = "string",
+                    ValueTypeId = "string"
+                }
+            }
+        };
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri.ToString().Contains("https://test.my.salesforce.com/services/data/v60.0/query?q=")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK)
+            { Content = new StringContent(JsonSerializer.Serialize(mockGetSchemaContent)) });
+
+
+
+        var mockUpdateJobStatusContent = new
+        {
+            id = "750QI000007QKdRYAW",
+            operation = "query",
+            @object = "ECCO_Consent__c",
+            createdById = "0057Y0000068CtmQAE",
+            createdDate = "2024-06-12T08:03:19.000+0000",
+            systemModstamp = "2024-06-12T08:03:41.000+0000",
+            state = "JobComplete",
+            concurrencyMode = "Parallel",
+            contentType = "CSV",
+            apiVersion = 60.0,
+            jobType = "V2Query",
+            lineEnding = "LF",
+            columnDelimiter = "COMMA",
+            numberRecordsProcessed = 449465,
+            retries = 0,
+            totalProcessingTime = 72011,
+            isPkChunkingSupported = true
+        };
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/data/v60.0/jobs/query/750QI000007QKdRYAW")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(new HttpResponseMessage(HttpStatusCode.OK)
+            { Content = new StringContent(JsonSerializer.Serialize(mockUpdateJobStatusContent)) });
+
+        var mockGetResultContent = @"""Id"",""Name""
+""1a"",""aa""
+""2b"",""bb""
+""3c"",""cc""
+""4d"",""dd""
+""5e"",""ee""";
+        var mockGetResultResponse = new HttpResponseMessage(HttpStatusCode.OK)
+        {
+            Content = new StringContent(mockGetResultContent)
+        };
+        mockGetResultResponse.Headers.Add("Sforce-Locator", "abcdefg");
+
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/data/v60.0/jobs/query/750QI000007QKdRYAW/results?maxRecords=5&")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(mockGetResultResponse);
+
+        var mockGetResultContent2 = @"""Id"",""Name""
+""6f"",""ff""
+""7g"",""gg""
+""8h"",""hh""
+""9i"",""ii""
+""10j"",""jj""";
+        var mockGetResultResponse2 = new HttpResponseMessage(HttpStatusCode.OK)
+        {
+            Content = new StringContent(mockGetResultContent2)
+        };
+        mockGetResultResponse2.Headers.Add("Sforce-Locator", "null");
+
+
+        this.mockHttp
+            .Setup(http =>
+                http.SendAsync(
+                    It.Is<HttpRequestMessage>(msg =>
+                        msg.RequestUri == new Uri("https://test.my.salesforce.com/services/data/v60.0/jobs/query/750QI000007QKdRYAW/results?maxRecords=5&locator=abcdefg")),
+                    It.IsAny<CancellationToken>()))
+            .ReturnsAsync(mockGetResultResponse2);
+
+        var source = SalesForceSource.Create(this.jobProvider, this.mockHttp.Object, "account", TimeSpan.FromSeconds(60));
+
+        var result = await Source.FromGraph(source)
+            .TakeWithin(TimeSpan.FromSeconds(10))
+            .RunWith(Sink.Seq<List<DataCell>>(), this.akkaFixture.Materializer);
+
+        Assert.Equal(10, result.Count);
+    }
+
+}