안녕하세요.
현재 .net Framework 프로젝트에 kafka를 바로 연동이 불가 하여 프로젝트를 하나 더 생성하여 .net Core 프로젝트를 참조하는 형태로 연동 테스트를 진행하고 있습니다.
Producer, Consumer 테스트 진행에는 이상이 없는데 Consumer 에서 계속 데이터를 수집하는 형식이 아니라 수집 시 추가 수집 데이터가 없으면 그만 멈추고 다시 제가 원할 때 호출하여 수집을 하는 형식으로 하고 싶은데 방법이 없을까요?
try
{
Console.WriteLine("--------------- Start Consumer ---------------");
var config = new ConsumerConfig
{
BootstrapServers = "broker01,broker02,broker03",
//ClientId = Dns.GetHostName(),
GroupId = "GroupId",
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = true
};
using(var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
consumer.Subscribe("topic");
CancellationTokenSource cts = new CancellationTokenSource();
try
{
while(!cts.IsCancellationRequested)
//while (true)
{
try
{
var consumerResult = consumer.Consume(cts.Token); // get consume result
if(consumerResult.Message == null)
//if (consumerResult.IsPartitionEOF)
{
Console.WriteLine("--------------- IsPartitionEOF ---------------");
consumer.Close();
break;
}
//consumer.Commit(consumerResult);
}
catch (ConsumeException cex)
{
Console.WriteLine("consumerException : " + cex.Message);
}
catch(Exception ex)
{
Console.WriteLine("consumer exception : " + ex.Message);
}
}
}
catch (OperationCanceledException ocex)
{
Console.WriteLine($"OperationCanceledException : {ocex.Message}");
consumer.Close();
}
catch(Exception ex)
{
Console.WriteLine($"exception : {ex.Message}");
consumer.Close();
}
finally
{
consumer.Close();
}
Console.WriteLine("--------------- End Consumer -------------- - ");
}
}
catch (Exception ex)
{
Console.WriteLine($"ex : {ex.Message}");
}