-
|
Disclaimer: I'm new to Kafka Streams so maybe this isn't related to this specific implementation. I'm able to forward messages from one topic to another and filter events as well. For example: using Confluent.Kafka;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
var config = new StreamConfig<StringSerDes, StringSerDes>();
config.ApplicationId = "test-app";
config.BootstrapServers = "pkc-n00kk.us-east-1.aws.confluent.cloud:9092";
config.SecurityProtocol = SecurityProtocol.SaslSsl;
config.SaslUsername = "<<SaslUsername>>";
config.SaslPassword = "<<SaslPassword>>";
config.SaslMechanism = SaslMechanism.Plain;
StreamBuilder builder = new StreamBuilder();
builder.Stream<string, string>("topic1")
.FilterNot((k, v) => v.Contains("test"))
.To("topic2");
Topology t = builder.Build();
KafkaStream stream = new KafkaStream(t, config);
Console.CancelKeyPress += (o, e) =>
{
stream.Dispose();
};
await stream.StartAsync();However, when I do any operation that involves tables or grouping/counting/aggregating the program exists after 5 seconds (I think this is related to the default value for For example: using Confluent.Kafka;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
var config = new StreamConfig<StringSerDes, StringSerDes>();
config.ApplicationId = "test-app";
config.BootstrapServers = "pkc-n00kk.us-east-1.aws.confluent.cloud:9092";
config.SecurityProtocol = SecurityProtocol.SaslSsl;
config.SaslUsername = "<<SaslUsername>>";
config.SaslPassword = "<<SaslPassword>>";
config.SaslMechanism = SaslMechanism.Plain;
StreamBuilder builder = new StreamBuilder();
builder.Stream<string, string>("topic1")
.GroupBy((k, v) => v)
.Count()
.ToStream()
.To("topic2");
Topology t = builder.Build();
KafkaStream stream = new KafkaStream(t, config);
Console.CancelKeyPress += (o, e) =>
{
stream.Dispose();
};
await stream.StartAsync();Or for example: using Confluent.Kafka;
using Streamiz.Kafka.Net;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
var config = new StreamConfig<StringSerDes, StringSerDes>();
config.ApplicationId = "test-app";
config.BootstrapServers = "pkc-n00kk.us-east-1.aws.confluent.cloud:9092";
config.SecurityProtocol = SecurityProtocol.SaslSsl;
config.SaslUsername = "<<SaslUsername>>";
config.SaslPassword = "<<SaslPassword>>";
config.SaslMechanism = SaslMechanism.Plain;
StreamBuilder builder = new StreamBuilder();
builder.Table("topic_9", InMemory<string, string>.As("test-store"))
.ToStream()
.To("topic_10");
Topology t = builder.Build();
KafkaStream stream = new KafkaStream(t, config);
Console.CancelKeyPress += (o, e) =>
{
stream.Dispose();
};
await stream.StartAsync();Notes: Any idea why this is happening and how to fix this? Thanks! |
Beta Was this translation helpful? Give feedback.
Answered by
LGouellec
Nov 21, 2022
Replies: 1 comment 6 replies
-
|
Hi @yornstei , Could you forward your logs when the issue happen please ? Kr |
Beta Was this translation helpful? Give feedback.
6 replies
-
|
Please update to config.Debug = "debug,security,broker,protocol";
config.Logger = LoggerFactory.Create(builder => {
builder.SetMinimumLevel(LogLevel.Debug);
builder.AddConsole(); // add your logger : console, log4net, whatever
}); |
Beta Was this translation helpful? Give feedback.
-
|
Updated to Removed the Thanks so much for the help!!! |
Beta Was this translation helpful? Give feedback.
-
|
@LGouellec from the error message I was able to see it had something todo with 'replication factor'. I added |
Beta Was this translation helpful? Give feedback.
-
|
@yornstei Because by default the replication factor used to create internal topic is 1, and with confluent cloud you can't create topic with rf = 1, so the topic creation failed and the streaming application can't start. Hope it's clear |
Beta Was this translation helpful? Give feedback.
-
|
@LGouellec Thanks for the help and clarification. |
Beta Was this translation helpful? Give feedback.
Answer selected by
yornstei
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Hi @yornstei ,
Could you forward your logs when the issue happen please ?
Kr