【发布时间】：2022-01-26 16:05:04
【问题描述】：
我正在研究 Akka.Streams.Kafka（参见页面 https://github.com/akkadotnet/Akka.Streams.Kafka 上的示例），我在创建消费者配置时遇到了问题
public class TestService : IHostedService
{
private ActorSystem _system;
private ActorMaterializer _materializer;
public async Task StartAsync(CancellationToken cancellationToken)
{
var actorsSetup = BootstrapSetup.Create()
.WithConfig(HoconLoader.ParseConfig("appsettings.conf"));
var kafkaConfig = HoconLoader.ParseConfig("kafka.conf");
_system = ActorSystem.Create("system", actorsSetup);
_materializer = _system.Materializer();
var consumerSettings = ConsumerSettings<Null, string>
.Create(kafkaConfig, null, Deserializers.Utf8)
.WithBootstrapServers("localhost:9092")
.WithDispatcher("")
.WithGroupId("group1"); // NULL-REF EXCEPTION HERE!
await KafkaConsumer.CommittableSource(consumerSettings, Subscriptions.Topics("TestTopic"))
.SelectAsync(1, async elem =>
{
await elem.CommitableOffset.Commit();
Console.WriteLine($"VAL:{elem.Record?.Value}");
return Done.Instance;
})
.RunWith(Sink.Ignore<Done>(), _materializer);
}
public async Task StopAsync(CancellationToken cancellationToken)
{
_materializer.Shutdown();
_system.Dispose();
}
}
尽管（看似）正确初始化，但当我尝试创建 ConsumerSettings 对象时出现“NUllReferenceException”。
System.NullReferenceException
HResult=0x80004003
Сообщение = Object reference not set to an instance of an object.
Источник = Akka.Streams.Kafka
Трассировка стека:
в Akka.Streams.Kafka.Settings.ConnectionCheckerSettings.Create(Config config)
我的 HOCON： appsettings.conf：
#####################################
# Akka Stream Reference Config File #
#####################################
akka {
stream {
# Default flow materializer settings
materializer {
# Initial size of buffers used in stream elements
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
initial-input-buffer-size = 4
# Maximum size of buffers used in stream elements
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
max-input-buffer-size = 16
# Fully qualified config path which holds the dispatcher configuration
# to be used by FlowMaterialiser when creating Actors.
# When this value is left empty, the default-dispatcher will be used.
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
dispatcher = ""
blocking-io-dispatcher = "akka.actor.default-blocking-io-dispatcher"
# Cleanup leaked publishers and subscribers when they are not used within a given
# deadline
subscription-timeout {
# when the subscription timeout is reached one of the following strategies on
# the "stale" publisher:
# cancel - cancel it (via `onError` or subscribing to the publisher and
# `cancel()`ing the subscription right away
# warn - log a warning statement about the stale element (then drop the
# reference to it)
# noop - do nothing (not recommended)
# Note: If you change this value also change the fallback value in StreamSubscriptionTimeoutSettings
mode = cancel
# time after which a subscriber / publisher is considered stale and eligible
# for cancelation (see `akka.stream.subscription-timeout.mode`)
# Note: If you change this value also change the fallback value in StreamSubscriptionTimeoutSettings
timeout = 5s
}
# Enable additional troubleshooting logging at DEBUG log level
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
debug-logging = off
# Maximum number of elements emitted in batch if downstream signals large demand
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
output-burst-limit = 1000
# Enable automatic fusing of all graphs that are run. For short-lived streams
# this may cause an initial runtime overhead, but most of the time fusing is
# desirable since it reduces the number of Actors that are created.
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
auto-fusing = on
# Those stream elements which have explicit buffers (like mapAsync, mapAsyncUnordered,
# buffer, flatMapMerge, Source.actorRef, Source.queue, etc.) will preallocate a fixed
# buffer upon stream materialization if the requested buffer size is less than this
# configuration parameter. The default is very high because failing early is better
# than failing under load.
#
# Buffers sized larger than this will dynamically grow/shrink and consume more memory
# per element than the fixed size buffers.
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
max-fixed-buffer-size = 1000000000
# Maximum number of sync messages that actor can process for stream to substream communication.
# Parameter allows to interrupt synchronous processing to get upsteam/downstream messages.
# Allows to accelerate message processing that happening withing same actor but keep system responsive.
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
sync-processing-limit = 1000
debug {
# Enables the fuzzing mode which increases the chance of race conditions
# by aggressively reordering events and making certain operations more
# concurrent than usual.
# This setting is for testing purposes, NEVER enable this in a production
# environment!
# To get the best results, try combining this setting with a throughput
# of 1 on the corresponding dispatchers.
# Note: If you change this value also change the fallback value in ActorMaterializerSettings
fuzzing-mode = off
}
stream-ref {
# Buffer of a SinkRef that is used to batch Request elements from the other side of the stream ref
#
# The buffer will be attempted to be filled eagerly even while the local stage did not request elements,
# because the delay of requesting over network boundaries is much higher.
buffer-capacity = 32
# Demand is signalled by sending a cumulative demand message ("requesting messages until the n-th sequence number)
# Using a cumulative demand model allows us to re-deliver the demand message in case of message loss (which should
# be very rare in any case, yet possible -- mostly under connection break-down and re-establishment).
#
# The semantics of handling and updating the demand however are in-line with what Reactive Streams dictates.
#
# In normal operation, demand is signalled in response to arriving elements, however if no new elements arrive
# within `demand-redelivery-interval` a re-delivery of the demand will be triggered, assuming that it may have gotten lost.
demand-redelivery-interval = 1 second
# Subscription timeout, during which the "remote side" MUST subscribe (materialize) the handed out stream ref.
# This timeout does not have to be very low in normal situations, since the remote side may also need to
# prepare things before it is ready to materialize the reference. However the timeout is needed to avoid leaking
# in-active streams which are never subscribed to.
subscription-timeout = 30 seconds
# In order to guard the receiving end of a stream ref from never terminating (since awaiting a Completion or Failed
# message) after / before a Terminated is seen, a special timeout is applied once Terminated is received by it.
# This allows us to terminate stream refs that have been targeted to other nodes which are Downed, and as such the
# other side of the stream ref would never send the "final" terminal message.
#
# The timeout specifically means the time between the Terminated signal being received and when the local SourceRef
# determines to fail itself, assuming there was message loss or a complete partition of the completion signal.
final-termination-signal-deadline = 2 seconds
}
}
# Deprecated, left here to not break Akka HTTP which refers to it
blocking-io-dispatcher = "akka.actor.default-blocking-io-dispatcher"
# Deprecated, will not be used unless user code refer to it, use 'akka.stream.materializer.blocking-io-dispatcher'
# instead, or if from code, prefer the 'ActorAttributes.IODispatcher' attribute
default-blocking-io-dispatcher = "akka.actor.default-blocking-io-dispatcher"
}
# configure overrides to ssl-configuration here (to be used by akka-streams, and akka-http – i.e. when serving https connections)
ssl-config {
protocol = "TLSv1"
}
actor {
serializers {
akka-stream-ref = "Akka.Streams.Serialization.StreamRefSerializer, Akka.Streams"
}
serialization-bindings {
"Akka.Streams.Implementation.StreamRef.SinkRefImpl, Akka.Streams" = akka-stream-ref
"Akka.Streams.Implementation.StreamRef.SourceRefImpl, Akka.Streams" = akka-stream-ref
"Akka.Streams.Implementation.StreamRef.IStreamRefsProtocol, Akka.Streams" = akka-stream-ref
}
serialization-identifiers {
"Akka.Streams.Serialization.StreamRefSerializer, Akka.Streams" = 30
}
}
}
kafka.hocon:
{
kafka-clients : {
enable : {
auto : {
commit : false
}
}
bootstrap : {
servers : "localhost:9092"
}
client : {
id : client-1
}
group : {
id : group-1
}
}
akka : {
kafka : {
consumer : {
poll-interval : 50ms
poll-timeout : 50ms
stop-timeout : 30s
commit-timeout : 15s
commit-time-warning : 1s
commit-refresh-interval : infinite
buffer-size : 128
use-dispatcher : akka.kafka.default-dispatcher
kafka-clients : {
enable : {
auto : {
commit : false
}
}
bootstrap : {
servers : "localhost:9092"
}
client : {
id : client-1
}
group : {
id : group-1
}
}
wait-close-partition : 500ms
position-timeout : 5s
offset-for-times-timeout : 5s
metadata-request-timeout : 5s
eos-draining-check-interval : 30ms
partition-handler-warning : 5s
}
}
}
}
【问题讨论】：
标签： c# apache-kafka stream queue akka.net