KafkaConsumerModule.java 文件源码

@Provides
@KafkaScope
ConsumerSchema.Consumer consumer(final IngestionManager ingestionManager) {
    // XXX: make target group configurable?
    final IngestionGroup ingestion = ingestionManager.useDefaultGroup();

    if (ingestion.isEmpty()) {
        throw new IllegalStateException("No backends are part of the ingestion group");
    }

    final ConsumerSchema.Depends d = DaggerConsumerSchema_Depends
        .builder()
        .primaryComponent(primary)
        .depends(depends)
        .dependsModule(new ConsumerSchema.DependsModule(ingestion))
        .build();

    final ConsumerSchema.Exposed exposed = schema.setup(d);
    return exposed.consumer();
}