Reformat lines; Bump Uploader InitialInterval backoff to 5 seconds

Author: vincepri

main.go

@@ -11,7 +11,7 @@ import (
 
 var (
 	// Version indicates the current version of the Kafka Achiever.
-	Version = "0.100"
+	Version = "0.101"
 )
 
 func init() {

uploader.go

@@ -93,7 +93,7 @@ func (w *Worker) upload(f *buffer.Flush) error {
 	// Create a new exponential backoff provider.
 	b := backoff.NewExponentialBackOff()
 	b.MaxElapsedTime = 5 * time.Minute
-	b.InitialInterval = 2 * time.Second
+	b.InitialInterval = 5 * time.Second
 
 	// S3 Uploader is wrapped in a backoff function with an exponential provider.
 	var s3Response *s3manager.UploadOutput

worker.go

@@ -85,41 +85,69 @@ type Worker struct {
 func (w *Worker) Init() {
 
 	// Define application flags using the `flag` package.
-	flag.Var(&w.brokers, "brokers", "Kafka brokers to connect to, as a comma separated list. (required)")
-	flag.Var(&w.topics, "topics", "Kafka topics to subscribe to.")
-	flag.StringVar(&w.topicWhitelist, "topic-whitelist", "", "An additional whitelist of topics to subscribe to.")
-	flag.StringVar(&w.topicBlacklist, "topic-blacklist", "", "An additional blacklist of topics, precedes the whitelist.")
-	flag.StringVar(&w.consumerGroup, "consumer-group", "", "Name of your kafka consumer group. (required)")
-	flag.StringVar(&w.consumerAutoOffsetReset, "consumer-auto-offset-reset", "oldest", "Kafka consumer group topic default offset.")
-	flag.Var(&w.consumerDelayStart, "consumer-delay-start", "Number of seconds to wait before starting consuming. (default \"0s\")")
+	flag.Var(&w.brokers, "brokers",
+		"Kafka brokers to connect to, as a comma separated list. (required)")
+	flag.Var(&w.topics, "topics",
+		"Kafka topics to subscribe to.")
+	flag.StringVar(&w.topicWhitelist, "topic-whitelist", "",
+		"An additional whitelist of topics to subscribe to.")
+	flag.StringVar(&w.topicBlacklist, "topic-blacklist", "",
+		"An additional blacklist of topics, precedes the whitelist.")
+	flag.StringVar(&w.consumerGroup, "consumer-group", "",
+		"Name of your kafka consumer group. (required)")
+	flag.StringVar(&w.consumerAutoOffsetReset, "consumer-auto-offset-reset", "oldest",
+		"Kafka consumer group topic default offset.")
+	flag.Var(&w.consumerDelayStart, "consumer-delay-start",
+		"Number of seconds to wait before starting consuming. (default \"0s\")")
 
 	// Define Buffer flags.
-	flag.Var(&w.bufferFlushInterval, "buffer-interval", "The duration when the buffer should close the file. (default \"15m\")")
-	flag.StringVar(&w.bufferFlushSize, "buffer-size", "100MiB", "The size in bytes when the buffer should close the file.")
-	flag.IntVar(&w.bufferQueueLength, "buffer-queue-length", 32, "The size of the queue to hold the files to be uploaded.")
-	flag.StringVar(&w.bufferLocation, "buffer-location", os.TempDir(), "The base folder where to store temporary files.")
-	flag.StringVar(&w.bufferMem, "buffer-mem", "8KB", "Amount of memory a buffer can use.")
+	flag.Var(&w.bufferFlushInterval, "buffer-interval",
+		"The duration when the buffer should close the file. (default \"15m\")")
+	flag.StringVar(&w.bufferFlushSize, "buffer-size", "100MiB",
+		"The size in bytes when the buffer should close the file.")
+	flag.IntVar(&w.bufferQueueLength, "buffer-queue-length", 32,
+		"The size of the queue to hold the files to be uploaded.")
+	flag.StringVar(&w.bufferLocation, "buffer-location", os.TempDir(),
+		"The base folder where to store temporary files.")
+	flag.StringVar(&w.bufferMem, "buffer-mem", "8KB",
+		"Amount of memory a buffer can use.")
 
 	// Define Partitioner flags.
-	flag.StringVar(&w.partitionerClass, "partitioner", "DefaultPartitioner", "The name of the partitioner to use.")
-	flag.StringVar(&w.partitionerFieldName, "partitioner-key", "", "Name of the JSON field to parse.")
-	flag.StringVar(&w.partitionerPathBaseFolder, "partitioner-path-folder", "backup", "The top level folder to prepend to the path used when partitioning files.")
-	flag.StringVar(&w.partitionerPathTopicNamePrefix, "partitioner-path-topic-prefix", "topic=", "A prefix to prepend to the path used when partitioning files.")
+	flag.StringVar(&w.partitionerClass, "partitioner", "DefaultPartitioner",
+		"The name of the partitioner to use.")
+	flag.StringVar(&w.partitionerFieldName, "partitioner-key", "",
+		"Name of the JSON field to parse.")
+	flag.StringVar(&w.partitionerPathBaseFolder,
+		"partitioner-path-folder", "backup",
+		"The top level folder to prepend to the path used when partitioning files.")
+	flag.StringVar(&w.partitionerPathTopicNamePrefix,
+		"partitioner-path-topic-prefix", "topic=",
+		"A prefix to prepend to the path used when partitioning files.")
 
 	// Define S3 flags.
 	flag.BoolVar(&w.s3enabled, "s3", false, "Enable S3 uploader.")
 	flag.StringVar(&w.s3region, "s3-region", "", "S3 Bucket Region.")
-	flag.StringVar(&w.s3bucket, "s3-bucket", "", "S3 Bucket where to upload files.")
-	flag.StringVar(&w.s3endpoint, "s3-endpoint", "", "S3 Bucket Endpoint to use for the client.")
-	flag.StringVar(&w.s3PartSize, "s3-part-size", "5MiB", "S3 Uploader part size.")
-	flag.IntVar(&w.s3Concurrency, "s3-concurrency", 5, "S3 Uploader part size.")
-	flag.BoolVar(&w.s3ForcePathStyle, "s3-force-path-style", false, "Enable to force the request to use path-style addressing on S3.")
-	flag.BoolVar(&w.s3ClientDebug, "s3-client-debug", false, "Enable to enable debug logging on S3 client.")
-	flag.StringVar(&w.s3NotificationTopic, "s3-notification-topic", "", "Kafka topic used to store uploaded S3 files.")
+	flag.StringVar(&w.s3bucket, "s3-bucket", "",
+		"S3 Bucket where to upload files.")
+	flag.StringVar(&w.s3endpoint, "s3-endpoint", "",
+		"S3 Bucket Endpoint to use for the client.")
+	flag.StringVar(&w.s3PartSize, "s3-part-size", "5MiB",
+		"S3 Uploader part size.")
+	flag.IntVar(&w.s3Concurrency, "s3-concurrency", 5,
+		"S3 Uploader part size.")
+	flag.BoolVar(&w.s3ForcePathStyle, "s3-force-path-style", false,
+		"Enable to force the request to use path-style addressing on S3.")
+	flag.BoolVar(&w.s3ClientDebug, "s3-client-debug", false,
+		"Enable to enable debug logging on S3 client.")
+	flag.StringVar(&w.s3NotificationTopic, "s3-notification-topic", "",
+		"Kafka topic used to store uploaded S3 files.")
 
 	// Define other flags.
-	flag.StringVar(&w.datadogHost, "datadog-host", flagutil.EnvOrDefault("DATADOG_HOST", "localhost:2585"), "The host where the datadog agents listens to.")
-	flag.StringVar(&w.statsdPrefix, "statsd-prefix", "kafka-archiver", "The name prefix for statsd metrics.")
+	flag.StringVar(&w.datadogHost, "datadog-host",
+		flagutil.EnvOrDefault("DATADOG_HOST", "localhost:2585"),
+		"The host where the datadog agents listens to.")
+	flag.StringVar(&w.statsdPrefix, "statsd-prefix", "kafka-archiver",
+		"The name prefix for statsd metrics.")
 
 	// Parse the flags.
 	flag.Parse()
@@ -199,8 +227,12 @@ func (w *Worker) Init() {
 		BufferSize:    int(bufferMemBytes),
 	}
 
-	glog.Infof("created buffer configuration flush-bytes=%d flush-interval=%v buffer-mem=%d",
-		int64(bufferFlushSize), w.bufferFlushInterval.Duration, int(bufferMemBytes))
+	glog.Infof(
+		"created buffer configuration flush-bytes=%v flush-interval=%v buffer-mem=%d",
+		int64(bufferFlushSize),
+		w.bufferFlushInterval.Duration,
+		int(bufferMemBytes),
+	)
 
 	// Create partitioner.
 	w.partitioner = partitioner.New(&partitioner.Config{
@@ -285,7 +317,12 @@ func (w *Worker) Init() {
 	}
 
 	// Create KAFKA consumer.
-	glog.Infof("creating consumer brokers=%+v topics=%+v topic-whitelist=%+v topic-blacklist=%+v", w.brokers, w.topics, w.topicWhitelist, w.topicBlacklist)
+	glog.Infof(
+		`creating consumer brokers=%+v topics=%+v 
+		topic-whitelist=%+v topic-blacklist=%+v`,
+		w.brokers, w.topics, w.topicWhitelist, w.topicBlacklist,
+	)
+
 	w.consumer, err = cluster.NewConsumer(w.brokers, w.consumerGroup, w.topics, config)
 	if err != nil {
 		glog.Fatal(err)
@@ -303,8 +340,9 @@ func (w *Worker) Init() {
 	w.wg.Wait()
 }
 
-// bufferListener listens on the channel that declares a file as flushed. Calls the upload function
-// to perform a backup to a cloud provider. Removes the file and marks the partition offset.
+// bufferListener listens on the channel that declares a file as flushed.
+// Calls the upload function to perform a backup to a cloud provider.
+// Removes the file and marks the partition offset.
 func (w *Worker) bufferListener() {
 	defer w.wg.Done()
 
@@ -318,7 +356,8 @@ func (w *Worker) bufferListener() {
 		}
 
 		w.consumer.MarkPartitionOffset(fl.Topic, fl.Partition, fl.LastOffset, "")
-		glog.V(2).Infof("marked offset: topic=%s partition=%d offset=%d", fl.Topic, fl.Partition, fl.LastOffset)
+		glog.V(2).Infof("marked offset: topic=%s partition=%d offset=%d",
+			fl.Topic, fl.Partition, fl.LastOffset)
 	}
 }