Rename max_batch_size to max_batch_count

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+## 0.41.0 - 2018-12-12
+
+### Changed
+
+- Renamed `max_batch_size` to `max_batch_count` for consistency.
+
 ## 0.40.2 - 2018-12-12
 
 ### Added

config/amqp.json

@@ -10,7 +10,7 @@
 		"amqp": {
 			"bindings_declare": [],
 			"consumer_tag": "benthos-consumer",
-			"max_batch_size": 1,
+			"max_batch_count": 1,
 			"prefetch_count": 10,
 			"prefetch_size": 0,
 			"queue": "benthos-queue",

config/amqp.yaml

@@ -9,7 +9,7 @@ input:
   amqp:
     bindings_declare: []
     consumer_tag: benthos-consumer
-    max_batch_size: 1
+    max_batch_count: 1
     prefetch_count: 10
     prefetch_size: 0
     queue: benthos-queue

config/env/README.md

@@ -54,7 +54,7 @@ HTTP_ROOT_PATH       = /benthos
 INPUTS                                      = 1
 INPUT_TYPE                                  = dynamic
 INPUT_AMQP_CONSUMER_TAG                     = benthos-consumer
-INPUT_AMQP_MAX_BATCH_SIZE                   = 1
+INPUT_AMQP_MAX_BATCH_COUNT                  = 1
 INPUT_AMQP_PREFETCH_COUNT                   = 10
 INPUT_AMQP_PREFETCH_SIZE                    = 0
 INPUT_AMQP_QUEUE                            = benthos-queue
@@ -117,7 +117,7 @@ INPUT_KAFKA_BALANCED_ADDRESSES              = localhost:9092
 INPUT_KAFKA_BALANCED_CLIENT_ID              = benthos_kafka_input
 INPUT_KAFKA_BALANCED_COMMIT_PERIOD_MS       = 1000
 INPUT_KAFKA_BALANCED_CONSUMER_GROUP         = benthos_consumer_group
-INPUT_KAFKA_BALANCED_MAX_BATCH_SIZE         = 1
+INPUT_KAFKA_BALANCED_MAX_BATCH_COUNT        = 1
 INPUT_KAFKA_BALANCED_START_FROM_OLDEST      = true
 INPUT_KAFKA_BALANCED_TARGET_VERSION         = 1.0.0
 INPUT_KAFKA_BALANCED_TLS_ENABLED            = false
@@ -127,7 +127,7 @@ INPUT_KAFKA_BALANCED_TOPICS                 = benthos_stream
 INPUT_KAFKA_CLIENT_ID                       = benthos_kafka_input
 INPUT_KAFKA_COMMIT_PERIOD_MS                = 1000
 INPUT_KAFKA_CONSUMER_GROUP                  = benthos_consumer_group
-INPUT_KAFKA_MAX_BATCH_SIZE                  = 1
+INPUT_KAFKA_MAX_BATCH_COUNT                 = 1
 INPUT_KAFKA_PARTITION                       = 0
 INPUT_KAFKA_START_FROM_OLDEST               = true
 INPUT_KAFKA_TARGET_VERSION                  = 1.0.0

config/env/default.yaml

@@ -10,7 +10,7 @@ input:
     inputs:
     - amqp:
         consumer_tag: ${INPUT_AMQP_CONSUMER_TAG:benthos-consumer}
-        max_batch_size: ${INPUT_AMQP_MAX_BATCH_SIZE:1}
+        max_batch_count: ${INPUT_AMQP_MAX_BATCH_COUNT:1}
         prefetch_count: ${INPUT_AMQP_PREFETCH_COUNT:10}
         prefetch_size: ${INPUT_AMQP_PREFETCH_SIZE:0}
         queue: ${INPUT_AMQP_QUEUE:benthos-queue}
@@ -90,7 +90,7 @@ input:
         client_id: ${INPUT_KAFKA_CLIENT_ID:benthos_kafka_input}
         commit_period_ms: ${INPUT_KAFKA_COMMIT_PERIOD_MS:1000}
         consumer_group: ${INPUT_KAFKA_CONSUMER_GROUP:benthos_consumer_group}
-        max_batch_size: ${INPUT_KAFKA_MAX_BATCH_SIZE:1}
+        max_batch_count: ${INPUT_KAFKA_MAX_BATCH_COUNT:1}
         partition: ${INPUT_KAFKA_PARTITION:0}
         start_from_oldest: ${INPUT_KAFKA_START_FROM_OLDEST:true}
         target_version: ${INPUT_KAFKA_TARGET_VERSION:1.0.0}
@@ -105,7 +105,7 @@ input:
         client_id: ${INPUT_KAFKA_BALANCED_CLIENT_ID:benthos_kafka_input}
         commit_period_ms: ${INPUT_KAFKA_BALANCED_COMMIT_PERIOD_MS:1000}
         consumer_group: ${INPUT_KAFKA_BALANCED_CONSUMER_GROUP:benthos_consumer_group}
-        max_batch_size: ${INPUT_KAFKA_BALANCED_MAX_BATCH_SIZE:1}
+        max_batch_count: ${INPUT_KAFKA_BALANCED_MAX_BATCH_COUNT:1}
         start_from_oldest: ${INPUT_KAFKA_BALANCED_START_FROM_OLDEST:true}
         target_version: ${INPUT_KAFKA_BALANCED_TARGET_VERSION:1.0.0}
         tls:

config/everything.yaml

@@ -15,7 +15,7 @@ input:
     consumer_tag: benthos-consumer
     prefetch_count: 10
     prefetch_size: 0
-    max_batch_size: 1
+    max_batch_count: 1
     tls:
       enabled: false
       root_cas_file: ""
@@ -99,7 +99,7 @@ input:
     partition: 0
     start_from_oldest: true
     target_version: 1.0.0
-    max_batch_size: 1
+    max_batch_count: 1
     tls:
       enabled: false
       root_cas_file: ""
@@ -115,7 +115,7 @@ input:
     - benthos_stream
     start_from_oldest: true
     target_version: 1.0.0
-    max_batch_size: 1
+    max_batch_count: 1
     tls:
       enabled: false
       root_cas_file: ""

config/kafka.json

@@ -14,7 +14,7 @@
 			"client_id": "benthos_kafka_input",
 			"commit_period_ms": 1000,
 			"consumer_group": "benthos_consumer_group",
-			"max_batch_size": 1,
+			"max_batch_count": 1,
 			"partition": 0,
 			"start_from_oldest": true,
 			"target_version": "1.0.0",

config/kafka.yaml

@@ -12,7 +12,7 @@ input:
     client_id: benthos_kafka_input
     commit_period_ms: 1000
     consumer_group: benthos_consumer_group
-    max_batch_size: 1
+    max_batch_count: 1
     partition: 0
     start_from_oldest: true
     target_version: 1.0.0

config/kafka_balanced.json

@@ -14,7 +14,7 @@
 			"client_id": "benthos_kafka_input",
 			"commit_period_ms": 1000,
 			"consumer_group": "benthos_consumer_group",
-			"max_batch_size": 1,
+			"max_batch_count": 1,
 			"start_from_oldest": true,
 			"target_version": "1.0.0",
 			"tls": {

config/kafka_balanced.yaml

@@ -12,7 +12,7 @@ input:
     client_id: benthos_kafka_input
     commit_period_ms: 1000
     consumer_group: benthos_consumer_group
-    max_batch_size: 1
+    max_batch_count: 1
     start_from_oldest: true
     target_version: 1.0.0
     tls:

docs/inputs/README.md

@@ -70,7 +70,7 @@ type: amqp
 amqp:
   bindings_declare: []
   consumer_tag: benthos-consumer
-  max_batch_size: 1
+  max_batch_count: 1
   prefetch_count: 10
   prefetch_size: 0
   queue: benthos-queue
@@ -88,7 +88,7 @@ amqp:
 Connects to an AMQP (0.91) queue. AMQP is a messaging protocol used by various
 message brokers, including RabbitMQ.
 
-The field `max_batch_size` specifies the maximum number of prefetched
+The field `max_batch_count` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the `split` processor.
 
@@ -412,7 +412,7 @@ kafka:
   client_id: benthos_kafka_input
   commit_period_ms: 1000
   consumer_group: benthos_consumer_group
-  max_batch_size: 1
+  max_batch_count: 1
   partition: 0
   start_from_oldest: true
   target_version: 1.0.0
@@ -429,7 +429,7 @@ consumer group (set via config). Only one partition per input is supported, if
 you wish to balance partitions across a consumer group look at the
 `kafka_balanced` input type instead.
 
-The field `max_batch_size` specifies the maximum number of prefetched
+The field `max_batch_count` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the `split` processor.
 
@@ -482,7 +482,7 @@ kafka_balanced:
   client_id: benthos_kafka_input
   commit_period_ms: 1000
   consumer_group: benthos_consumer_group
-  max_batch_size: 1
+  max_batch_count: 1
   start_from_oldest: true
   target_version: 1.0.0
   tls:
@@ -498,7 +498,7 @@ Connects to a kafka (0.9+) server. Offsets are managed within kafka as per the
 consumer group (set via config), and partitions are automatically balanced
 across any members of the consumer group.
 
-The field `max_batch_size` specifies the maximum number of prefetched
+The field `max_batch_count` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the `split` processor.
 

lib/input/amqp.go

@@ -36,7 +36,7 @@ func init() {
 Connects to an AMQP (0.91) queue. AMQP is a messaging protocol used by various
 message brokers, including RabbitMQ.
 
-The field ` + "`max_batch_size`" + ` specifies the maximum number of prefetched
+The field ` + "`max_batch_count`" + ` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the ` + "`split`" + ` processor.
 

lib/input/kafka.go

@@ -39,7 +39,7 @@ consumer group (set via config). Only one partition per input is supported, if
 you wish to balance partitions across a consumer group look at the
 ` + "`kafka_balanced`" + ` input type instead.
 
-The field ` + "`max_batch_size`" + ` specifies the maximum number of prefetched
+The field ` + "`max_batch_count`" + ` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the ` + "`split`" + ` processor.
 

lib/input/kafka_balanced.go

@@ -38,7 +38,7 @@ Connects to a kafka (0.9+) server. Offsets are managed within kafka as per the
 consumer group (set via config), and partitions are automatically balanced
 across any members of the consumer group.
 
-The field ` + "`max_batch_size`" + ` specifies the maximum number of prefetched
+The field ` + "`max_batch_count`" + ` specifies the maximum number of prefetched
 messages to be batched together. When more than one message is batched they can
 be split into individual messages with the ` + "`split`" + ` processor.
 

lib/input/reader/amqp.go

@@ -61,7 +61,7 @@ type AMQPConfig struct {
 	ConsumerTag     string                 `json:"consumer_tag" yaml:"consumer_tag"`
 	PrefetchCount   int                    `json:"prefetch_count" yaml:"prefetch_count"`
 	PrefetchSize    int                    `json:"prefetch_size" yaml:"prefetch_size"`
-	MaxBatchSize    int                    `json:"max_batch_size" yaml:"max_batch_size"`
+	MaxBatchCount   int                    `json:"max_batch_count" yaml:"max_batch_count"`
 	TLS             btls.Config            `json:"tls" yaml:"tls"`
 }
 
@@ -78,7 +78,7 @@ func NewAMQPConfig() AMQPConfig {
 		PrefetchCount:   10,
 		PrefetchSize:    0,
 		TLS:             btls.NewConfig(),
-		MaxBatchSize:    1,
+		MaxBatchCount:   1,
 		BindingsDeclare: []AMQPBindingConfig{},
 	}
 }
@@ -335,7 +335,7 @@ func (a *AMQP) Read() (types.Message, error) {
 	addPart(data)
 
 batchLoop:
-	for i := 1; i < a.conf.MaxBatchSize; i++ {
+	for i := 1; i < a.conf.MaxBatchCount; i++ {
 		select {
 		case data, open = <-c:
 			if !open {

lib/input/reader/amqp_test.go

@@ -199,7 +199,7 @@ func testAMQPBatch(url string, t *testing.T) {
 	conf := NewAMQPConfig()
 	conf.URL = url
 	conf.QueueDeclare.Enabled = true
-	conf.MaxBatchSize = 10
+	conf.MaxBatchCount = 10
 	conf.BindingsDeclare = append(conf.BindingsDeclare, AMQPBindingConfig{
 		Exchange:   exchange,
 		RoutingKey: key,
@@ -256,8 +256,8 @@ func testAMQPBatch(url string, t *testing.T) {
 				ContentType:     "application/octet-stream",
 				ContentEncoding: "",
 				Body:            []byte(testStr),
-				DeliveryMode:    amqp.Transient, // 1=non-persistent, 2=persistent
-				Priority:        0,              // 0-9
+				DeliveryMode:    1, // 1=non-persistent, 2=persistent
+				Priority:        0, // 0-9
 			}); pErr != nil {
 				t.Error(pErr)
 			}

lib/input/reader/kafka.go

@@ -47,7 +47,7 @@ type KafkaConfig struct {
 	Partition       int32       `json:"partition" yaml:"partition"`
 	StartFromOldest bool        `json:"start_from_oldest" yaml:"start_from_oldest"`
 	TargetVersion   string      `json:"target_version" yaml:"target_version"`
-	MaxBatchSize    int         `json:"max_batch_size" yaml:"max_batch_size"`
+	MaxBatchCount   int         `json:"max_batch_count" yaml:"max_batch_count"`
 	TLS             btls.Config `json:"tls" yaml:"tls"`
 }
 
@@ -62,7 +62,7 @@ func NewKafkaConfig() KafkaConfig {
 		Partition:       0,
 		StartFromOldest: true,
 		TargetVersion:   sarama.V1_0_0_0.String(),
-		MaxBatchSize:    1,
+		MaxBatchCount:   1,
 		TLS:             btls.NewConfig(),
 	}
 }
@@ -299,7 +299,7 @@ func (k *Kafka) Read() (types.Message, error) {
 	addPart(data)
 
 batchLoop:
-	for i := 1; i < k.conf.MaxBatchSize; i++ {
+	for i := 1; i < k.conf.MaxBatchCount; i++ {
 		select {
 		case data, open = <-partConsumer.Messages():
 			if !open {

lib/input/reader/kafka_balanced.go

@@ -47,7 +47,7 @@ type KafkaBalancedConfig struct {
 	Topics          []string    `json:"topics" yaml:"topics"`
 	StartFromOldest bool        `json:"start_from_oldest" yaml:"start_from_oldest"`
 	TargetVersion   string      `json:"target_version" yaml:"target_version"`
-	MaxBatchSize    int         `json:"max_batch_size" yaml:"max_batch_size"`
+	MaxBatchCount   int         `json:"max_batch_count" yaml:"max_batch_count"`
 	TLS             btls.Config `json:"tls" yaml:"tls"`
 }
 
@@ -61,7 +61,7 @@ func NewKafkaBalancedConfig() KafkaBalancedConfig {
 		Topics:          []string{"benthos_stream"},
 		StartFromOldest: true,
 		TargetVersion:   sarama.V1_0_0_0.String(),
-		MaxBatchSize:    1,
+		MaxBatchCount:   1,
 		TLS:             btls.NewConfig(),
 	}
 }
@@ -265,7 +265,7 @@ func (k *KafkaBalanced) Read() (types.Message, error) {
 	addPart(data)
 
 batchLoop:
-	for i := 1; i < k.conf.MaxBatchSize; i++ {
+	for i := 1; i < k.conf.MaxBatchCount; i++ {
 		select {
 		case data, open = <-consumer.Messages():
 			if !open {