mirror of
https://github.com/1f349/dendrite.git
synced 2024-11-08 18:16:59 +00:00
Report which component failed to consume (#1375)
This commit is contained in:
parent
6d79f04354
commit
89c772fb78
@ -48,6 +48,7 @@ func NewOutputRoomEventConsumer(
|
||||
workerStates []types.ApplicationServiceWorkerState,
|
||||
) *OutputRoomEventConsumer {
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "appservice/roomserver",
|
||||
Topic: cfg.Global.Kafka.TopicFor(config.TopicOutputRoomEvent),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: appserviceDB,
|
||||
|
@ -36,6 +36,7 @@ type OutputRoomEventConsumer struct {
|
||||
|
||||
func NewOutputRoomEventConsumer(topicName string, kafkaConsumer sarama.Consumer, store storage.Database, acls *acls.ServerACLs) *OutputRoomEventConsumer {
|
||||
consumer := &internal.ContinualConsumer{
|
||||
ComponentName: "currentstateserver/roomserver",
|
||||
Topic: topicName,
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -50,11 +50,13 @@ func NewOutputEDUConsumer(
|
||||
) *OutputEDUConsumer {
|
||||
c := &OutputEDUConsumer{
|
||||
typingConsumer: &internal.ContinualConsumer{
|
||||
ComponentName: "eduserver/typing",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
},
|
||||
sendToDeviceConsumer: &internal.ContinualConsumer{
|
||||
ComponentName: "eduserver/sendtodevice",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -49,6 +49,7 @@ func NewKeyChangeConsumer(
|
||||
) *KeyChangeConsumer {
|
||||
c := &KeyChangeConsumer{
|
||||
consumer: &internal.ContinualConsumer{
|
||||
ComponentName: "federationsender/keychange",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputKeyChangeEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -48,6 +48,7 @@ func NewOutputRoomEventConsumer(
|
||||
rsAPI api.RoomserverInternalAPI,
|
||||
) *OutputRoomEventConsumer {
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "federationsender/roomserver",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -33,6 +33,7 @@ type PartitionStorer interface {
|
||||
// A ContinualConsumer continually consumes logs even across restarts. It requires a PartitionStorer to
|
||||
// remember the offset it reached.
|
||||
type ContinualConsumer struct {
|
||||
ComponentName string
|
||||
// The kafkaesque topic to consume events from.
|
||||
// This is the name used in kafka to identify the stream to consume events from.
|
||||
Topic string
|
||||
@ -111,7 +112,7 @@ func (c *ContinualConsumer) consumePartition(pc sarama.PartitionConsumer) {
|
||||
msgErr := c.ProcessMessage(message)
|
||||
// Advance our position in the stream so that we will start at the right position after a restart.
|
||||
if err := c.PartitionStore.SetPartitionOffset(context.TODO(), c.Topic, message.Partition, message.Offset); err != nil {
|
||||
panic(fmt.Errorf("the ContinualConsumer failed to SetPartitionOffset: %w", err))
|
||||
panic(fmt.Errorf("the ContinualConsumer in %q failed to SetPartitionOffset: %w", c.ComponentName, err))
|
||||
}
|
||||
// Shutdown if we were told to do so.
|
||||
if msgErr == ErrShutdown {
|
||||
|
@ -44,6 +44,7 @@ func NewOutputClientDataConsumer(
|
||||
) *OutputClientDataConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "syncapi/clientapi",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputClientData)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -48,6 +48,7 @@ func NewOutputSendToDeviceEventConsumer(
|
||||
) *OutputSendToDeviceEventConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "syncapi/eduserver/sendtodevice",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -44,6 +44,7 @@ func NewOutputTypingEventConsumer(
|
||||
) *OutputTypingEventConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "syncapi/eduserver/typing",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -56,6 +56,7 @@ func NewOutputKeyChangeEventConsumer(
|
||||
) *OutputKeyChangeEventConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "syncapi/keychange",
|
||||
Topic: topic,
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
@ -49,6 +49,7 @@ func NewOutputRoomEventConsumer(
|
||||
) *OutputRoomEventConsumer {
|
||||
|
||||
consumer := internal.ContinualConsumer{
|
||||
ComponentName: "syncapi/roomserver",
|
||||
Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputRoomEvent)),
|
||||
Consumer: kafkaConsumer,
|
||||
PartitionStore: store,
|
||||
|
Loading…
Reference in New Issue
Block a user