[ksqlDb.RestApi.Client]: fixed Source table can't be queried #87

Author: tomasfabian

Tests/ksqlDB.RestApi.Client.Tests/KSql/RestApi/KSqlDbRestApiClientTests.cs

@@ -1,4 +1,5 @@
 using System.Linq.Expressions;
+using System.Net;
 using System.Text;
 using FluentAssertions;
 using ksqlDb.RestApi.Client.Infrastructure.Logging;
@@ -701,4 +702,52 @@ public async Task AssertSchemaExistsAsync_ReturnsFalse()
     //Assert
     response[0].Exists.Should().BeFalse();
   }
+
+  [Test]
+  public async Task CreateSourceTableAsync()
+  {
+    //Arrange
+    CreateHttpMocks(@"[{""@type"":""tables""}]");
+
+    var creationMetadata = new EntityCreationMetadata("moviesByTitle", 1)
+    {
+      Replicas = 1,
+      Partitions = 1,
+      ShouldPluralizeEntityName = true,
+    };
+
+    //Act
+    var response = await ClassUnderTest.CreateSourceTableAsync<Movie>(creationMetadata);
+
+    //Assert
+    var expectedContent = GetExpectedContent(@"CREATE SOURCE TABLE Movies (\r\n\tTitle VARCHAR,\r\n\tId INT PRIMARY KEY,\r\n\tRelease_Year INT\r\n) WITH ( KAFKA_TOPIC=\u0027moviesByTitle\u0027, VALUE_FORMAT=\u0027Json\u0027, PARTITIONS=\u00271\u0027, REPLICAS=\u00271\u0027 );".ReplaceLineEndings());
+
+    VerifySendAsync(expectedContent);
+
+    response.StatusCode.Should().Be(HttpStatusCode.OK);
+  }
+
+  [Test]
+  public async Task CreateSourceStreamAsync()
+  {
+    //Arrange
+    CreateHttpMocks(@"[{""@type"":""streams""}]");
+
+    var creationMetadata = new EntityCreationMetadata("moviesByTitle", 1)
+    {
+      Replicas = 1,
+      Partitions = 1,
+      ShouldPluralizeEntityName = true,
+    };
+
+    //Act
+    var response = await ClassUnderTest.CreateSourceStreamAsync<Movie>(creationMetadata);
+
+    //Assert
+    var expectedContent = GetExpectedContent(@"CREATE SOURCE STREAM Movies (\r\n\tTitle VARCHAR,\r\n\tId INT KEY,\r\n\tRelease_Year INT\r\n) WITH ( KAFKA_TOPIC=\u0027moviesByTitle\u0027, VALUE_FORMAT=\u0027Json\u0027, PARTITIONS=\u00271\u0027, REPLICAS=\u00271\u0027 );".ReplaceLineEndings());
+
+    VerifySendAsync(expectedContent);
+
+    response.StatusCode.Should().Be(HttpStatusCode.OK);
+  }
 }

ksqlDb.RestApi.Client/ChangeLog.md

@@ -1,5 +1,10 @@
 # ksqlDB.RestApi.Client
 
+# 6.2.1
+
+## BugFix
+- Source table can't be queried #87 - fixed `KSqlDbRestApiClient.CreateSourceTableAsync` and `KSqlDbRestApiClient.CreateSourceStreamAsync`
+
 # 6.2.0
 
 1. **KSqlDbRestApiClient Constructor Update**:

ksqlDb.RestApi.Client/KSql/RestApi/KSqlDbRestApiClient.cs

@@ -281,11 +281,15 @@ public Task<HttpResponseMessage> CreateSourceStreamAsync<T>(EntityCreationMetada
     ArgumentNullException.ThrowIfNull(creationMetadata);
 #endif
 
+    creationMetadata = creationMetadata with
+    {
+      IsReadOnly = true
+    };
+
     if (creationMetadata.ShouldPluralizeEntityName == null)
       creationMetadata = creationMetadata with
       {
         ShouldPluralizeEntityName = clientOptions.ShouldPluralizeFromItemName,
-        IsReadOnly = true
       };
 
     var ksql = statementGenerator.CreateStream<T>(creationMetadata, ifNotExists);
@@ -333,11 +337,15 @@ public Task<HttpResponseMessage> CreateSourceTableAsync<T>(EntityCreationMetadat
     ArgumentNullException.ThrowIfNull(creationMetadata);
 #endif
 
+    creationMetadata = creationMetadata with
+    {
+      IsReadOnly = true
+    };
+
     if (creationMetadata.ShouldPluralizeEntityName == null)
       creationMetadata = creationMetadata with
       {
         ShouldPluralizeEntityName = clientOptions.ShouldPluralizeFromItemName,
-        IsReadOnly = true
       };
 
     var ksql = statementGenerator.CreateTable<T>(creationMetadata, ifNotExists);

ksqlDb.RestApi.Client/ksqlDb.RestApi.Client.csproj

@@ -15,7 +15,7 @@
             Documentation for the library can be found at https://github.com/tomasfabian/ksqlDB.RestApi.Client-DotNet/blob/main/README.md.
         </Description>
         <PackageTags>ksql ksqlDB LINQ .NET csharp push query</PackageTags>
-        <Version>6.2.0</Version>
+        <Version>6.2.1</Version>
         <AssemblyVersion>6.2.0.0</AssemblyVersion>
         <LangVersion>12.0</LangVersion>
         <ImplicitUsings>enable</ImplicitUsings>