-
Notifications
You must be signed in to change notification settings - Fork 59
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
chore: logging fix with bigquery optimisations (#155)
- Loading branch information
Showing
7 changed files
with
199 additions
and
85 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,29 +1,97 @@ | ||
| package io.odpf.firehose.sink; | ||
|
|
||
|
|
||
| import io.odpf.firehose.config.KafkaConsumerConfig; | ||
| import io.odpf.firehose.consumer.kafka.OffsetManager; | ||
| import io.odpf.firehose.exception.ConfigurationException; | ||
| import io.odpf.firehose.metrics.Instrumentation; | ||
| import io.odpf.firehose.metrics.StatsDReporter; | ||
| import io.odpf.firehose.sink.bigquery.BigQuerySinkFactory; | ||
| import io.odpf.firehose.sink.blob.BlobSinkFactory; | ||
| import io.odpf.firehose.sink.elasticsearch.EsSinkFactory; | ||
| import io.odpf.firehose.sink.grpc.GrpcSinkFactory; | ||
| import io.odpf.firehose.sink.http.HttpSinkFactory; | ||
| import io.odpf.firehose.sink.influxdb.InfluxSinkFactory; | ||
| import io.odpf.firehose.sink.jdbc.JdbcSinkFactory; | ||
| import io.odpf.firehose.sink.log.LogSinkFactory; | ||
| import io.odpf.firehose.sink.mongodb.MongoSinkFactory; | ||
| import io.odpf.firehose.sink.prometheus.PromSinkFactory; | ||
| import io.odpf.firehose.sink.redis.RedisSinkFactory; | ||
| import io.odpf.stencil.client.StencilClient; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| /** | ||
| * Interface to create the sink. | ||
| * | ||
| * Any sink {@see Sink} can be created by a class implementing this interface. | ||
| * The consumer framework would reflectively instantiate this factory | ||
| * using the configurations supplied and invoke {@see #create(Map<String, String> configuration, StatsDClient client)} | ||
| * to obtain the sink implementation. | ||
| * | ||
| */ | ||
| public interface SinkFactory { | ||
| public class SinkFactory { | ||
| private final KafkaConsumerConfig kafkaConsumerConfig; | ||
| private final StatsDReporter statsDReporter; | ||
| private final Instrumentation instrumentation; | ||
| private final StencilClient stencilClient; | ||
| private final OffsetManager offsetManager; | ||
| private BigQuerySinkFactory bigQuerySinkFactory; | ||
| private final Map<String, String> config = System.getenv(); | ||
|
|
||
| public SinkFactory(KafkaConsumerConfig kafkaConsumerConfig, | ||
| StatsDReporter statsDReporter, | ||
| StencilClient stencilClient, | ||
| OffsetManager offsetManager) { | ||
| instrumentation = new Instrumentation(statsDReporter, SinkFactory.class); | ||
| this.kafkaConsumerConfig = kafkaConsumerConfig; | ||
| this.statsDReporter = statsDReporter; | ||
| this.stencilClient = stencilClient; | ||
| this.offsetManager = offsetManager; | ||
| } | ||
|
|
||
| /** | ||
| * method to create the sink from configuration supplied. | ||
| * | ||
| * @param configuration key/value configuration supplied as a map | ||
| * @param client {@see StatsDClient} | ||
| * @param stencilClient {@see StencilClient} | ||
| * @return instance of sink to which messages consumed from kafka can be forwarded to. {@see Sink} | ||
| * Initialization method for all the sinks. | ||
| */ | ||
| Sink create(Map<String, String> configuration, StatsDReporter client, StencilClient stencilClient); | ||
| public void init() { | ||
| switch (this.kafkaConsumerConfig.getSinkType()) { | ||
| case JDBC: | ||
| case HTTP: | ||
| case INFLUXDB: | ||
| case LOG: | ||
| case ELASTICSEARCH: | ||
| case REDIS: | ||
| case GRPC: | ||
| case PROMETHEUS: | ||
| case BLOB: | ||
| case MONGODB: | ||
| return; | ||
| case BIGQUERY: | ||
| bigQuerySinkFactory = new BigQuerySinkFactory(config, statsDReporter); | ||
| bigQuerySinkFactory.init(); | ||
| return; | ||
| default: | ||
| throw new ConfigurationException("Invalid Firehose SINK_TYPE"); | ||
| } | ||
| } | ||
|
|
||
| public Sink getSink() { | ||
| instrumentation.logInfo("Sink Type: {}", kafkaConsumerConfig.getSinkType().toString()); | ||
| switch (kafkaConsumerConfig.getSinkType()) { | ||
| case JDBC: | ||
| return JdbcSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case HTTP: | ||
| return HttpSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case INFLUXDB: | ||
| return InfluxSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case LOG: | ||
| return LogSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case ELASTICSEARCH: | ||
| return EsSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case REDIS: | ||
| return RedisSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case GRPC: | ||
| return GrpcSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case PROMETHEUS: | ||
| return PromSinkFactory.create(config, statsDReporter, stencilClient); | ||
| case BLOB: | ||
| return BlobSinkFactory.create(config, offsetManager, statsDReporter, stencilClient); | ||
| case BIGQUERY: | ||
| return bigQuerySinkFactory.create(); | ||
| case MONGODB: | ||
| return MongoSinkFactory.create(config, statsDReporter, stencilClient); | ||
| default: | ||
| throw new ConfigurationException("Invalid Firehose SINK_TYPE"); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.