Skip to content

Commit 7566330

Browse files
devstressdevstress
committed
Init
1 parent 5a10e1e commit 7566330

15 files changed

Lines changed: 789 additions & 44 deletions

File tree

‎BackPressureExample/BackPressure.IntegrationTests/KafkaTestBase.cs‎

Lines changed: 72 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -26,22 +26,82 @@ public abstract class KafkaTestBase : IAsyncDisposable
2626
public virtual async Task OneTimeSetUp()
2727
{
2828
var cancellationToken = TestContext.CurrentContext.CancellationToken;
29-
var appHost = await DistributedApplicationTestingBuilder.CreateAsync<Projects.BackPressure_AppHost>(cancellationToken);
30-
var app = await appHost.BuildAsync(cancellationToken).WaitAsync(defaultTimeout, cancellationToken);
31-
await app.StartAsync(cancellationToken).WaitAsync(defaultTimeout, cancellationToken);
3229

33-
await app.ResourceNotifications
34-
.WaitForResourceHealthyAsync("kafka", cancellationToken)
35-
.WaitAsync(defaultTimeout, cancellationToken);
30+
// Extended startup timeout (prior 60s) to reduce flakiness under cold Docker starts.
31+
var extendedTimeout = TimeSpan.FromSeconds(120);
3632

37-
KafkaConnectionString = await app.GetConnectionStringAsync("kafka");
38-
39-
TestContext.WriteLine($"✅ Kafka connection string: {KafkaConnectionString}");
33+
int attempt = 0;
34+
const int maxAttempts = 3;
35+
DistributedApplication? app = null;
4036

41-
42-
await WaitForKafkaReadyAsync(KafkaConnectionString!, TimeSpan.FromSeconds(30), cancellationToken);
37+
while (attempt < maxAttempts)
38+
{
39+
attempt++;
40+
try
41+
{
42+
TestContext.WriteLine($"🟡 [AppHost] Attempt {attempt}/{maxAttempts} creating test host (Timeout={extendedTimeout.TotalSeconds}s)");
43+
var swCreate = Stopwatch.StartNew();
44+
45+
// TEMP: Revert to existing AppHost until Runner project is added to BackPressureExample.sln
46+
// var appHostBuilder = await DistributedApplicationTestingBuilder.CreateAsync<Projects.BackPressure_Runner>(cancellationToken);
47+
var appHostBuilder = await DistributedApplicationTestingBuilder.CreateAsync<Projects.BackPressure_AppHost>(cancellationToken);
48+
49+
TestContext.WriteLine("🟡 [AppHost] Building distributed application...");
50+
var buildSw = Stopwatch.StartNew();
51+
app = await appHostBuilder.BuildAsync(cancellationToken).WaitAsync(extendedTimeout, cancellationToken);
52+
buildSw.Stop();
53+
TestContext.WriteLine($"✅ [AppHost] Build complete in {buildSw.Elapsed.TotalSeconds:F1}s");
54+
55+
TestContext.WriteLine("🟡 [AppHost] Starting distributed application...");
56+
var startSw = Stopwatch.StartNew();
57+
await app.StartAsync(cancellationToken).WaitAsync(extendedTimeout, cancellationToken);
58+
startSw.Stop();
59+
TestContext.WriteLine($"✅ [AppHost] Start complete in {startSw.Elapsed.TotalSeconds:F1}s");
60+
61+
swCreate.Stop();
62+
TestContext.WriteLine($"✅ [AppHost] Infrastructure up in {swCreate.Elapsed.TotalSeconds:F1}s (attempt {attempt})");
63+
64+
TestContext.WriteLine("🟡 [Health] Awaiting kafka resource healthy notification...");
65+
var healthSw = Stopwatch.StartNew();
66+
await app.ResourceNotifications
67+
.WaitForResourceHealthyAsync("kafka", cancellationToken)
68+
.WaitAsync(extendedTimeout, cancellationToken);
69+
healthSw.Stop();
70+
TestContext.WriteLine($"✅ [Health] Kafka reported healthy in {healthSw.Elapsed.TotalSeconds:F1}s");
71+
72+
KafkaConnectionString = await app.GetConnectionStringAsync("kafka");
73+
TestContext.WriteLine($"✅ Kafka connection string: {KafkaConnectionString}");
74+
75+
var readinessTimeout = TimeSpan.FromSeconds(45);
76+
TestContext.WriteLine($"🟡 [KafkaReady] Probing cluster readiness (timeout {readinessTimeout.TotalSeconds}s)...");
77+
await WaitForKafkaReadyAsync(KafkaConnectionString!, readinessTimeout, cancellationToken);
78+
79+
await SetupKafkaClientsAsync();
80+
81+
TestContext.WriteLine("✅ [Setup] Infrastructure & clients initialized successfully");
82+
AppHost = app;
83+
break; // success
84+
}
85+
catch (Exception ex) when (attempt < maxAttempts)
86+
{
87+
TestContext.WriteLine($"⚠️ [Retry] Attempt {attempt} failed to initialize infrastructure: {ex.GetType().Name} - {ex.Message}");
88+
if (ex is TimeoutException)
89+
{
90+
TestContext.WriteLine("ℹ️ [Retry] Detected timeout; increasing grace period before next attempt...");
91+
}
92+
var backoffMs = attempt * 3000;
93+
await Task.Delay(backoffMs, cancellationToken);
94+
}
95+
catch
96+
{
97+
throw;
98+
}
99+
}
43100

44-
await SetupKafkaClientsAsync();
101+
if (app == null)
102+
{
103+
throw new TimeoutException("Failed to initialize distributed application after retries.");
104+
}
45105

46106
TestContext.WriteLine(
47107
$"🟢 Infrastructure initialized: Kafka={KafkaConnectionString}, " +

‎FlinkDotNet/Flink.JobBuilder/Services/FlinkJobGatewayService.cs‎

Lines changed: 50 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -63,37 +63,78 @@ public async Task<JobSubmissionResult> SubmitJobAsync(JobDefinition jobDefinitio
6363
return JobSubmissionResult.CreateFailure(jobDefinition.Metadata.JobId, msg);
6464
}
6565

66+
// Serialize IR (capture diagnostics about polymorphic discriminator presence)
6667
var json = JsonSerializer.Serialize(jobDefinition, _jsonOptions);
68+
var hasDiscriminatorToken = json.Contains("\"type\"", StringComparison.Ordinal);
69+
var firstSnippet = json.Length > 500 ? json[..500] + "...(truncated)" : json;
70+
_logger?.LogInformation(
71+
"Job {JobId} JSON serialized (length={Length}, hasDiscriminatorToken={HasType}). Snippet: {Snippet}",
72+
jobDefinition.Metadata.JobId,
73+
json.Length,
74+
hasDiscriminatorToken,
75+
firstSnippet);
76+
77+
// Additional focused check: count discriminator occurrences for debugging polymorphic binding
78+
if (_logger != null)
79+
{
80+
var typeCount = 0;
81+
var idx = 0;
82+
while ((idx = json.IndexOf("\"type\"", idx, StringComparison.Ordinal)) >= 0)
83+
{
84+
typeCount++;
85+
idx += 6;
86+
}
87+
_logger.LogDebug("Job {JobId} discriminator occurrences: {TypeCount}", jobDefinition.Metadata.JobId, typeCount);
88+
}
89+
6790
var content = new StringContent(json, Encoding.UTF8, "application/json");
6891

6992
var response = await ExecuteWithRetryAsync(async () =>
7093
{
7194
return await _httpClient.PostAsync("/api/v1/jobs/submit", content, cancellationToken);
7295
});
7396

97+
var rawResponse = await response.Content.ReadAsStringAsync(cancellationToken);
98+
var responseSnippet = rawResponse.Length > 600 ? rawResponse[..600] + "...(truncated)" : rawResponse;
99+
74100
if (response.IsSuccessStatusCode)
75101
{
76-
var responseContent = await response.Content.ReadAsStringAsync(cancellationToken);
77-
var result = JsonSerializer.Deserialize<JobSubmissionResult>(responseContent, _jsonOptions);
78-
102+
JobSubmissionResult? result = null;
103+
try
104+
{
105+
result = JsonSerializer.Deserialize<JobSubmissionResult>(rawResponse, _jsonOptions);
106+
}
107+
catch (Exception ex)
108+
{
109+
_logger?.LogError(ex, "Deserialization of JobSubmissionResult failed for Job {JobId}. Raw response snippet: {Snippet}",
110+
jobDefinition.Metadata.JobId, responseSnippet);
111+
}
112+
79113
if (result != null)
80114
{
81115
result.SubmittedAt = DateTime.UtcNow;
82-
_logger?.LogInformation("Job {JobId} submitted successfully. Flink Job ID: {FlinkJobId}",
83-
jobDefinition.Metadata.JobId, result.FlinkJobId);
116+
_logger?.LogInformation("Job {JobId} submitted successfully. Flink Job ID: {FlinkJobId}. Raw response snippet: {Snippet}",
117+
jobDefinition.Metadata.JobId, result.FlinkJobId, responseSnippet);
84118
return result;
85119
}
120+
121+
_logger?.LogWarning("Job {JobId} submission success status but null result. Raw response snippet: {Snippet}",
122+
jobDefinition.Metadata.JobId, responseSnippet);
123+
}
124+
else
125+
{
126+
_logger?.LogWarning("Job {JobId} submission failed HTTP {Status}. Raw response snippet: {Snippet}",
127+
jobDefinition.Metadata.JobId, response.StatusCode, responseSnippet);
86128
}
87129

88-
var errorContent = await response.Content.ReadAsStringAsync(cancellationToken);
89-
_logger?.LogError("Failed to submit job {JobId}. Status: {StatusCode}, Error: {Error}",
90-
jobDefinition.Metadata.JobId, response.StatusCode, errorContent);
130+
_logger?.LogError("Failed to submit job {JobId}. Status: {StatusCode}",
131+
jobDefinition.Metadata.JobId, response.StatusCode);
91132

92133
return new JobSubmissionResult
93134
{
94135
JobId = jobDefinition.Metadata.JobId,
95136
Success = false,
96-
ErrorMessage = $"HTTP {response.StatusCode}: {errorContent}",
137+
ErrorMessage = $"HTTP {response.StatusCode}: {responseSnippet}",
97138
SubmittedAt = DateTime.UtcNow
98139
};
99140
}

‎FlinkDotNet/Flink.JobGateway/Program.cs‎

Lines changed: 92 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,39 @@
1+
using System.Text;
2+
using System.Text.Json;
3+
using System.Text.Json.Serialization.Metadata;
14
using Flink.JobGateway.Services;
5+
using Microsoft.AspNetCore.Mvc;
6+
using Microsoft.AspNetCore.Mvc.Filters;
27
using Microsoft.OpenApi.Models;
38

49
var builder = WebApplication.CreateBuilder(args);
510

6-
// Add services to the container
7-
builder.Services.AddControllers();
11+
// Add controllers with JSON + ModelState logging
12+
builder.Services
13+
.AddControllers(options =>
14+
{
15+
options.Filters.Add<ModelStateLoggingFilter>();
16+
})
17+
.AddJsonOptions(o =>
18+
{
19+
o.JsonSerializerOptions.PropertyNamingPolicy = JsonNamingPolicy.CamelCase;
20+
o.JsonSerializerOptions.WriteIndented = false;
21+
// Insert default resolver so interface polymorphic attributes are honored consistently
22+
o.JsonSerializerOptions.TypeInfoResolverChain.Insert(0, new DefaultJsonTypeInfoResolver());
23+
});
24+
825
builder.Services.AddEndpointsApiExplorer();
926
builder.Services.AddSwaggerGen(c =>
1027
{
11-
c.SwaggerDoc("v1", new OpenApiInfo
12-
{
28+
c.SwaggerDoc("v1", new OpenApiInfo
29+
{
1330
Title = "Flink Job Gateway API",
1431
Version = "v1",
1532
Description = "REST API for submitting and managing Apache Flink jobs from .NET applications"
1633
});
1734
});
1835

19-
// Add API versioning
36+
// API versioning
2037
builder.Services.AddApiVersioning(options =>
2138
{
2239
options.AssumeDefaultVersionWhenUnspecified = true;
@@ -29,10 +46,10 @@
2946
options.SubstituteApiVersionInUrl = true;
3047
});
3148

32-
// Register services
49+
// Services
3350
builder.Services.AddHttpClient<IFlinkJobManager, FlinkJobManager>();
3451

35-
// Configure logging
52+
// Logging
3653
builder.Services.AddLogging(loggingBuilder =>
3754
{
3855
loggingBuilder.AddConsole();
@@ -41,22 +58,87 @@
4158

4259
var app = builder.Build();
4360

44-
// Configure the HTTP request pipeline
61+
// Diagnostic middleware: capture raw body + 400 responses for /api/v1/jobs/submit
62+
app.Use(async (ctx, next) =>
63+
{
64+
var isSubmit = ctx.Request.Path.Equals("/api/v1/jobs/submit", StringComparison.OrdinalIgnoreCase);
65+
if (isSubmit)
66+
{
67+
try
68+
{
69+
ctx.Request.EnableBuffering();
70+
using var reader = new StreamReader(ctx.Request.Body, Encoding.UTF8, leaveOpen: true);
71+
var raw = await reader.ReadToEndAsync();
72+
ctx.Request.Body.Position = 0;
73+
var log = ctx.RequestServices.GetRequiredService<ILoggerFactory>().CreateLogger("JobSubmitRawBody");
74+
log.LogInformation("Raw job submission body: {Body}", raw);
75+
}
76+
catch (Exception ex)
77+
{
78+
var log = ctx.RequestServices.GetRequiredService<ILoggerFactory>().CreateLogger("JobSubmitRawBody");
79+
log.LogWarning(ex, "Failed to read raw submission body.");
80+
}
81+
}
82+
83+
var originalBody = ctx.Response.Body;
84+
using var mem = new MemoryStream();
85+
ctx.Response.Body = mem;
86+
87+
await next();
88+
89+
if (isSubmit && ctx.Response.StatusCode == 400)
90+
{
91+
mem.Position = 0;
92+
var bodyText = await new StreamReader(mem).ReadToEndAsync();
93+
var log = ctx.RequestServices.GetRequiredService<ILoggerFactory>().CreateLogger("JobSubmitModelState");
94+
log.LogWarning("Job submission returned 400. Response body: {Body}", bodyText);
95+
mem.Position = 0;
96+
}
97+
98+
await mem.CopyToAsync(originalBody);
99+
ctx.Response.Body = originalBody;
100+
});
101+
102+
// Pipeline
45103
if (app.Environment.IsDevelopment())
46104
{
47105
app.UseSwagger();
48106
app.UseSwaggerUI(c =>
49107
{
50108
c.SwaggerEndpoint("/swagger/v1/swagger.json", "Flink Job Gateway API v1");
51-
c.RoutePrefix = string.Empty; // Make Swagger UI the default page
109+
c.RoutePrefix = string.Empty;
52110
});
53111
}
54112

55113
app.UseAuthorization();
56114
app.MapControllers();
57115

58-
// Health check endpoint
116+
// Health endpoints
59117
app.MapGet("/health", () => Results.Ok("OK"));
60118
app.MapGet("/api/v1/health", () => Results.Ok(new { status = "OK", timestamp = DateTime.UtcNow }));
61119

62120
await app.RunAsync();
121+
122+
/// <summary>
123+
/// Logs ModelState validation errors (including polymorphic binding issues).
124+
/// </summary>
125+
internal sealed class ModelStateLoggingFilter : IActionFilter
126+
{
127+
private readonly ILogger<ModelStateLoggingFilter> _logger;
128+
public ModelStateLoggingFilter(ILogger<ModelStateLoggingFilter> logger) => _logger = logger;
129+
130+
public void OnActionExecuting(ActionExecutingContext context)
131+
{
132+
if (!context.ModelState.IsValid)
133+
{
134+
var errors = context.ModelState
135+
.Where(kv => kv.Value?.Errors.Count > 0)
136+
.Select(kv => $"{kv.Key}:{string.Join("|", kv.Value!.Errors.Select(e => e.ErrorMessage))}");
137+
_logger.LogWarning("ModelState invalid for {Path}. Errors: {Errors}",
138+
context.HttpContext.Request.Path,
139+
string.Join("; ", errors));
140+
}
141+
}
142+
143+
public void OnActionExecuted(ActionExecutedContext context) { }
144+
}

‎FlinkDotNet/Flink.JobGateway/Services/FlinkJobManager.cs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -538,7 +538,10 @@ private static void ValidateBasicProperties(JobDefinition jobDefinition, List<st
538538
if (jobDefinition.Source == null)
539539
errors.Add("Job source is required");
540540

541-
if (jobDefinition.Sink == null)
541+
// Allow sink-less SQL jobs (SQL statements may define sinks).
542+
// This aligns gateway validation with client-side JobDefinitionValidator.
543+
var isSqlJob = jobDefinition.Source is SqlSourceDefinition;
544+
if (jobDefinition.Sink == null && !isSqlJob)
542545
errors.Add("Job sink is required");
543546
}
544547

0 commit comments

Comments
 (0)