kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jjko...@apache.org
Subject git commit: KAFKA-991; Rename config queue size to queue bytes in hadoop producer; patched by Swapnil Ghike, reviewed by Joel Koshy.
Date Fri, 02 Aug 2013 00:12:06 GMT
Updated Branches:
  refs/heads/0.8 8edd3e630 -> b6479e8ad


KAFKA-991; Rename config queue size to queue bytes in hadoop producer; patched by Swapnil
Ghike, reviewed by Joel Koshy.


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/b6479e8a
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/b6479e8a
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/b6479e8a

Branch: refs/heads/0.8
Commit: b6479e8adca6be8e363d289be29dc755e0655550
Parents: 8edd3e6
Author: Joel Koshy <jjkoshy@gmail.com>
Authored: Thu Aug 1 17:11:58 2013 -0700
Committer: Joel Koshy <jjkoshy@gmail.com>
Committed: Thu Aug 1 17:11:58 2013 -0700

----------------------------------------------------------------------
 contrib/hadoop-producer/README.md               |  4 ++--
 .../kafka/bridge/hadoop/KafkaOutputFormat.java  |  7 +++----
 .../kafka/bridge/hadoop/KafkaRecordWriter.java  | 20 +++++++++++---------
 3 files changed, 16 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/b6479e8a/contrib/hadoop-producer/README.md
----------------------------------------------------------------------
diff --git a/contrib/hadoop-producer/README.md b/contrib/hadoop-producer/README.md
index 547c1ef..a5bef73 100644
--- a/contrib/hadoop-producer/README.md
+++ b/contrib/hadoop-producer/README.md
@@ -77,8 +77,8 @@ turned off by the OutputFormat.
 What can I tune?
 ----------------
 
-* kafka.output.queue.size: Bytes to queue in memory before pushing to the Kafka
-  producer (i.e., the batch size). Default is 10*1024*1024 (10MB).
+* kafka.output.queue.bytes: Bytes to queue in memory before pushing to the Kafka
+  producer (i.e., the batch size). Default is 1,000,000 (1 million) bytes.
 
 Any of Kafka's producer parameters can be changed by prefixing them with
 "kafka.output" in one's job configuration. For example, to change the

http://git-wip-us.apache.org/repos/asf/kafka/blob/b6479e8a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaOutputFormat.java
----------------------------------------------------------------------
diff --git a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaOutputFormat.java
b/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaOutputFormat.java
index 709a609..417b4b3 100644
--- a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaOutputFormat.java
+++ b/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaOutputFormat.java
@@ -44,7 +44,7 @@ public class KafkaOutputFormat<K, V> extends OutputFormat<K, V>
    *  We set the default to a million bytes so that the server will not reject the batch
of messages
    *  with a MessageSizeTooLargeException. The actual size will be smaller after compression.
    */
-  public static final int KAFKA_QUEUE_SIZE = 1000000;
+  public static final int KAFKA_QUEUE_BYTES = 1000000;
 
   public static final String KAFKA_CONFIG_PREFIX = "kafka.output";
   private static final Map<String, String> kafkaConfigMap;
@@ -119,8 +119,7 @@ public class KafkaOutputFormat<K, V> extends OutputFormat<K,
V>
     }
 
     // KafkaOutputFormat specific parameters
-    final int queueSize = job.getInt(KAFKA_CONFIG_PREFIX + ".queue.size", KAFKA_QUEUE_SIZE);
-    job.setInt(KAFKA_CONFIG_PREFIX + ".queue.size", queueSize);
+    final int queueBytes = job.getInt(KAFKA_CONFIG_PREFIX + ".queue.bytes", KAFKA_QUEUE_BYTES);
 
     if (uri.getScheme().equals("kafka")) {
       // using the direct broker list
@@ -140,6 +139,6 @@ public class KafkaOutputFormat<K, V> extends OutputFormat<K,
V>
       throw new KafkaException("missing scheme from kafka uri (must be kafka://)");
 
     Producer<Object, byte[]> producer = new Producer<Object, byte[]>(new ProducerConfig(props));
-    return new KafkaRecordWriter<K, V>(producer, topic, queueSize);
+    return new KafkaRecordWriter<K, V>(producer, topic, queueBytes);
   }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/b6479e8a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaRecordWriter.java
----------------------------------------------------------------------
diff --git a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaRecordWriter.java
b/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaRecordWriter.java
index 6eea635..72c088d 100644
--- a/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaRecordWriter.java
+++ b/contrib/hadoop-producer/src/main/java/kafka/bridge/hadoop/KafkaRecordWriter.java
@@ -32,14 +32,14 @@ public class KafkaRecordWriter<K,V> extends RecordWriter<K,V>
   protected String topic;
 
   protected List<KeyedMessage<Object, byte[]>> msgList = new LinkedList<KeyedMessage<Object,
byte[]>>();
-  protected int totalSize = 0;
-  protected int queueSize;
+  protected int totalBytes = 0;
+  protected int queueBytes;
 
-  public KafkaRecordWriter(Producer<Object, byte[]> producer, String topic, int queueSize)
+  public KafkaRecordWriter(Producer<Object, byte[]> producer, String topic, int queueBytes)
   {
     this.producer = producer;
     this.topic = topic;
-    this.queueSize = queueSize;
+    this.queueBytes = queueBytes;
   }
 
   protected void sendMsgList() throws IOException
@@ -52,7 +52,7 @@ public class KafkaRecordWriter<K,V> extends RecordWriter<K,V>
         throw new IOException(e);           // all Kafka exceptions become IOExceptions
       }
       msgList.clear();
-      totalSize = 0;
+      totalBytes = 0;
     }
   }
 
@@ -69,12 +69,14 @@ public class KafkaRecordWriter<K,V> extends RecordWriter<K,V>
     else
       throw new IllegalArgumentException("KafkaRecordWriter expects byte array value to publish");
 
-    msgList.add(new KeyedMessage<Object, byte[]>(this.topic, key, valBytes));
-    totalSize += valBytes.length;
-
     // MultiProducerRequest only supports sending up to Short.MAX_VALUE messages in one batch
-    if (totalSize > queueSize || msgList.size() >= Short.MAX_VALUE)
+    // If the new message is going to make the message list tip over 1 million bytes, send
the
+    // message list now.
+    if ((totalBytes + valBytes.length) > queueBytes || msgList.size() >= Short.MAX_VALUE)
       sendMsgList();
+
+    msgList.add(new KeyedMessage<Object, byte[]>(this.topic, key, valBytes));
+    totalBytes += valBytes.length;
   }
 
   @Override


Mime
View raw message