Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Fixed command-line issue with boolean options

  • Loading branch information...
commit 871c8b96cdbc92ec177bc776e6fd9d80fbb0e741 1 parent 9508f9d
@xstevens xstevens authored
View
7 src/main/java/com/mozilla/bagheera/consumer/KafkaConsumer.java
@@ -76,7 +76,8 @@
protected ValidationPipeline validationPipeline;
protected Meter consumed;
-
+ protected Meter invalidMessageMeter;
+
public KafkaConsumer(String topic, Properties props) {
this(topic, props, DEFAULT_NUM_THREADS);
}
@@ -91,6 +92,7 @@ public KafkaConsumer(String topic, Properties props, int numThreads) {
streams = consumerConnector.createMessageStreamsByFilter(new Whitelist(topic), numThreads);
consumed = Metrics.newMeter(new MetricName("bagheera", "consumer", topic + ".consumed"), "messages", TimeUnit.SECONDS);
+ invalidMessageMeter = Metrics.newMeter(new MetricName("bagheera", "consumer", topic + ".invalid"), "messages", TimeUnit.SECONDS);
}
public void setSinkFactory(KeyValueSinkFactory sinkFactory) {
@@ -158,13 +160,14 @@ public Void call() {
sink.store(bmsg.getId(), bmsg.getPayload().toByteArray());
}
} else {
+ invalidMessageMeter.mark();
LOG.warn("Invalid payload for namespace: " + bmsg.getNamespace());
}
} else if (bmsg.getOperation() == Operation.DELETE &&
bmsg.hasId()) {
sink.delete(bmsg.getId());
}
- consumed.mark();
+ consumed.mark();
}
} catch (InvalidProtocolBufferException e) {
LOG.error("Invalid protocol buffer in data stream", e);
View
2  src/main/java/com/mozilla/bagheera/consumer/KafkaHBaseConsumer.java
@@ -67,7 +67,7 @@ public static void main(String[] args) {
sinkConfig.setString("hbasesink.hbase.tablename", cmd.getOptionValue("table"));
sinkConfig.setString("hbasesink.hbase.column.family", cmd.getOptionValue("family", "data"));
sinkConfig.setString("hbasesink.hbase.column.qualifier", cmd.getOptionValue("qualifier", "json"));
- sinkConfig.setBoolean("hbasesink.hbase.rowkey.prefixdate", Boolean.parseBoolean(cmd.getOptionValue("prefixdate", "false")));
+ sinkConfig.setBoolean("hbasesink.hbase.rowkey.prefixdate", cmd.hasOption("prefixdate"));
KeyValueSinkFactory sinkFactory = KeyValueSinkFactory.getInstance(HBaseSink.class, sinkConfig);
sh.addLast(sinkFactory);
View
2  src/main/java/com/mozilla/bagheera/consumer/KafkaSequenceFileConsumer.java
@@ -64,7 +64,7 @@ public static void main(String[] args) {
sinkConfig.setString("hdfssink.hdfs.basedir.path", cmd.getOptionValue("output", "/bagheera"));
sinkConfig.setString("hdfssink.hdfs.date.format", cmd.getOptionValue("dateformat", "yyyy-MM-dd"));
sinkConfig.setLong("hdfssink.hdfs.max.filesize", Long.parseLong(cmd.getOptionValue("filesize", "536870912")));
- sinkConfig.setBoolean("hdfssink.hdfs.usebytes", Boolean.parseBoolean(cmd.getOptionValue("usebytes", "false")));
+ sinkConfig.setBoolean("hdfssink.hdfs.usebytes", cmd.hasOption("usebytes"));
KeyValueSinkFactory sinkFactory = KeyValueSinkFactory.getInstance(SequenceFileSink.class, sinkConfig);
sh.addLast(sinkFactory);
View
17 src/main/java/com/mozilla/bagheera/util/IdUtil.java
@@ -22,6 +22,7 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.text.SimpleDateFormat;
+import java.util.Calendar;
import java.util.Date;
import org.apache.commons.lang.StringUtils;
@@ -97,4 +98,20 @@ public static String hbaseShellId(String id, Date d) throws IOException {
return sb.toString();
}
+ public static String hbaseShellId(byte[] idBytes) throws IOException {
+ StringBuilder sb = new StringBuilder("\"\\x");
+ sb.append(String.format("%02x", idBytes[0]));
+ sb.append((new String(idBytes)).substring(1));
+ sb.append("\"");
+ return sb.toString();
+ }
+
+ public static void main(String[] args) throws IOException {
+ String id = "06b87727-2946-4b1e-b654-8bafb60bb995";
+ long timestamp = 1360493115891L;
+ Calendar cal = Calendar.getInstance();
+ //cal.set(2013, Calendar.FEBRUARY, 10);
+ cal.setTimeInMillis(timestamp);
+ System.out.println(IdUtil.hbaseShellId(id, cal.getTime()));
+ }
}
Please sign in to comment.
Something went wrong with that request. Please try again.