SQL Server & Kafka

void Foo()
{
    using(var transaction = connection.BeginTransaction())
    { }

    using(var transaction = connection.BeginTransaction())
    { }
    
    Bar();
}

void Bar()
{
    using(var transaction = connection.BeginTransaction())
    { }
}

CREATE TABLE storage
(
    storageId INT IDENTITY(1,1) PRIMARY KEY,
    imageId INT NOT NULL,
    x INT NOT NULL,
    y INT NOT NULL,
    UNIQUE(imageId, x, y)
);

CREATE TABLE rgb
(    
    storageId INT NOT NULL FOREIGN KEY REFERENCES storage(storageId) PRIMARY KEY, 
    red INT NOT NULL,
    green INT NOT NULL,
    blue INT NOT NULL    
);

SELECT * FROM fn_dblog(null, null)

SELECT is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 1 AND [name] = 'storage'

IF @@ROWCOUNT = 0
    EXEC sys.sp_cdc_enable_table 
        @source_schema = N'dbo', 
        @source_name   = N'storage', 
        @role_name     = NULL 

SELECT is_tracked_by_cdc FROM sys.tables WHERE is_tracked_by_cdc = 1 AND [name] = 'rgb'

IF @@ROWCOUNT = 0
    EXEC sys.sp_cdc_enable_table 
        @source_schema = N'dbo', 
        @source_name   = N'rgb', 
        @role_name     = NULL 

GO

SELECT name, is_tracked_by_cdc FROM sys.tables
SELECT capture_instance FROM cdc.change_tables
SELECT * from sys.dm_cdc_log_scan_sessions ORDER BY start_time DESC

EXECUTE sys.sp_cdc_change_job   
    @job_type = N'capture',  
	@continuous = 1,
	@maxtrans = 1000,
    @maxscans = 100,  
    @pollinginterval = 1;

EXEC sys.sp_cdc_stop_job @job_type = 'capture';
EXEC sys.sp_cdc_start_job @job_type = 'capture';

private static async Task Iterate(FileInfo image, Func<IDbConnection, IDbTransaction, int, int, Task> callback)
{
    var bitmap = new Bitmap(image.FullName);
 
    using (IDbConnection connection = new SqlConnection(@"XXXX"))
    {
        connection.Open();
 
        for (int y = 0; y < bitmap.Height; y++)
        {
            for (int x = 0; x < bitmap.Width; x++)
            {
                // Small transaction = "more work" for CDC / Kafka
                using (var transation = connection.BeginTransaction())
                {
                    await callback(connection, transation, x, y);
 
                    transation.Commit();
                }
            }
        }
    }
}

private static async Task WriteImageData(MyImage imageFile)
{
    var bitmap = new Bitmap(imageFile.File.FullName);
 
    await Iterate(imageFile.File, async (connection, transation, x, y) =>
    {
        var color = bitmap.GetPixel(x, y);
 
        var imageSql = @"INSERT INTO [dbo].[storage] (imageId, x, y) VALUES(@ImageId, @X, @Y); SELECT CAST(SCOPE_IDENTITY() AS INT);";
 
        var storageId = await connection.QueryAsync<int>(imageSql, new { ImageId = imageFile.ImageId, x, y }, transaction: transation);
 
        var pixelSql = @"INSERT INTO [dbo].[rgb] (storageId, red, green, blue) VALUES(@StorageId, @Red, @Green, @Blue);";
 
        await connection.ExecuteAsync(pixelSql, new { StorageId = storageId, Red = color.R, Green = color.G, Blue = color.B }, transaction: transation);
    });
}

private static void HandleTimerElapsed(object sender, ElapsedEventArgs e)
{
    Timer.Enabled = false;
 
    var storageQuery = @"DECLARE @min_lsn binary(10), @from_lsn binary(10), @to_lsn binary(10);                 
        SET @to_lsn = sys.fn_cdc_get_max_lsn(); 
        SET @min_lsn = sys.fn_cdc_get_min_lsn ('dbo_storage'); 
        SET @from_lsn = sys.fn_cdc_increment_lsn(@SavedLsn); 
 
        SELECT @to_lsn AS toLsn;
         
        IF(@from_lsn < @min_lsn)
            SET @from_lsn = @min_lsn;
 
        IF (@from_lsn > @to_lsn) 
            SET @from_lsn = @to_lsn;               
         
        SELECT [__$start_lsn] AS startLsn, [__$seqval] AS seqVal, [__$operation] AS operation, [storageId], [imageId], [x], [y] FROM cdc.fn_cdc_get_all_changes_dbo_storage(@from_lsn, @to_lsn, 'all');";
 
    var rgbQuery = @"DECLARE @min_lsn binary(10), @from_lsn binary(10), @to_lsn binary(10);                 
        SET @to_lsn = sys.fn_cdc_get_max_lsn(); 
        SET @min_lsn = sys.fn_cdc_get_min_lsn ('dbo_rgb'); 
        SET @from_lsn = sys.fn_cdc_increment_lsn(@SavedLsn); 
 
        SELECT @to_lsn AS toLsn;
         
        IF(@from_lsn < @min_lsn)
            SET @from_lsn = @min_lsn;
 
        IF (@from_lsn > @to_lsn) 
            SET @from_lsn = @to_lsn;               
         
        SELECT [__$start_lsn] AS startLsn, [__$seqval] AS seqVal, [__$operation] AS operation, [storageId], [red], [green], [blue] FROM cdc.fn_cdc_get_all_changes_dbo_rgb(@from_lsn, @to_lsn, 'all');";
 
    using (IDbConnection connection = new SqlConnection(@"XXX"))
    {
        connection.Open();
 
        using (var reader = connection.QueryMultiple(storageQuery, new { SavedLsn = lastStorageToLsn }))
        {
            var range = reader.ReadSingle<dynamic>();
            var changes = reader.Read<dynamic>();
            var changesAsRecords = changes.Select(i => new StorageRecord(i.startLsn, i.seqVal, i.operation, i.storageId, i.imageId, i.x, i.y));
 
            lastStorageToLsn = range.toLsn;
 
            StorageRecords.AddRange(changesAsRecords);
        }
 
        using (var reader = connection.QueryMultiple(rgbQuery, new { SavedLsn = lastRgbToLsn }))
        {
            var range = reader.ReadSingle<dynamic>();
            var changes = reader.Read<dynamic>();
            var changesAsRecords = changes.Select(i => new RgbRecord(i.startLsn, i.seqVal, i.operation, i.storageId, i.red, i.green, i.blue));
 
            lastRgbToLsn = range.toLsn;
 
            RgbRecords.AddRange(changesAsRecords);
        }
    }
 
    var mergedResult = MergeSortedLists(StorageRecords, RgbRecords);
 
    var config = new ProducerConfig
    {
        BootstrapServers = "localhost:9092",
        EnableIdempotence = true
    };
 
    Action<DeliveryReportResult<string, byte[]>> handler = r =>
    {
        if (r.Error.IsError)
        {
            Console.WriteLine(r.Error.Reason);
        }
    };
 
    var policy = Policy
        .Handle<KafkaException>(exception => exception.Error.Code == ErrorCode.Local_QueueFull)
        .WaitAndRetryForever(i => TimeSpan.FromSeconds(5));
 
    using (var p = new Producer<string, byte[]>(config))
    {
        foreach (var item in mergedResult)
        {
            var key = string.Empty;
 
            if (item is StorageRecord)
            {
                var storageRecord = (StorageRecord)item;
 
                key = $"storage/{storageRecord.ImageId}/{storageRecord.X}/{storageRecord.Y}";
            }
 
            if (item is RgbRecord)
            {
                var rgbRecord = (RgbRecord)item;
 
                key = $"rgb/{rgbRecord.StorageId}";
            }
 
            policy.Execute(() =>
            {
                p.BeginProduce("test", new Message<string, byte[]>
                {
                    Key = key,
                    Value = Serialize(item)
                }, handler);
            });
 
            Interlocked.Increment(ref processedItems);
        }
 
        p.Flush(TimeSpan.FromSeconds(10));
    }
 
    Console.WriteLine("Iteration finished!");
 
    StorageRecords.Clear();
    RgbRecords.Clear();
 
    Timer.Enabled = true;
}

public static IList<Record> MergeSortedLists(IList<Record> list1, IList<Record> list2)
{
    var resultList = new List<Record>(list1.Count + list2.Count);
    var comparer = new ByteComparer();

    int list1Index = 0, list2Index = 0;

    for (int i = 0; i < list1.Count + list2.Count; i++)
    {
        if (list2Index >= list2.Count)
        {
            if (list1.Any())
            {
                resultList.Add(list1[list1Index++]);
            }
        }

        if (list1Index >= list1.Count)
        {
            if (list2.Any())
            {
                resultList.Add(list2[list2Index++]);
            }
        }

        if (list1Index < list1.Count && list2Index < list2.Count)
        {
            if (comparer.Compare(list1[list1Index].Lsn, list2[list2Index].Lsn) < 0)
            {
                resultList.Add(list1[list1Index++]);
            }
            else if (comparer.Compare(list1[list1Index].Lsn, list2[list2Index].Lsn) > 0)
            {
                resultList.Add(list2[list2Index++]);
            }
            else
            {
                if (comparer.Compare(list1[list1Index].Sequence, list2[list2Index].Sequence) < 0)
                {
                    resultList.Add(list1[list1Index++]);
                }
                else
                {
                    resultList.Add(list2[list2Index++]);
                }
            }
        }
    }

    return resultList;
}