Merge remote-tracking branch 'origin/fluo-425'
diff --git a/docs/applications.md b/docs/applications.md
index 7aaedc3..5685381 100644
--- a/docs/applications.md
+++ b/docs/applications.md
@@ -100,7 +100,27 @@
To create an observer, follow these steps:
-1. Create a class that extends [AbstractObserver].
+1. Create a class that extends [AbstractObserver] like the example below. Please use [slf4j] for
+ any logging in observers as [slf4j] supports multiple logging implementations. This is
+ necessary as Fluo applications have a hard requirement on [logback] when running in YARN.
+
+ ```java
+ public class InvertObserver extends AbstractObserver {
+
+ @Override
+ public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
+ // read value
+ Bytes value = tx.get(row, col);
+ // invert row and value
+ tx.set(value, new Column("inv", "data"), row);
+ }
+
+ @Override
+ public ObservedColumn getObservedColumn() {
+ return new ObservedColumn(new Column("obs", "data"), NotificationType.STRONG);
+ }
+ }
+ ```
2. Build a jar containing this class and include this jar in the `lib/` directory of your Fluo
application.
3. Configure your Fluo instance to use this observer by modifying the Observer section of
@@ -179,3 +199,5 @@
[fluo.properties]: ../modules/distribution/src/main/config/fluo.properties
[API]: https://fluo.apache.org/apidocs/
[metrics]: metrics.md
+[slf4j]: http://www.slf4j.org/
+[logback]: http://logback.qos.ch/
diff --git a/docs/install.md b/docs/install.md
index 18f4380..818249f 100644
--- a/docs/install.md
+++ b/docs/install.md
@@ -157,10 +157,37 @@
fluo info myapp
-You can also use `yarn application -list` to check the status of your Fluo instance in YARN. Logs
-are viewable within YARN.
+You can also use `yarn application -list` to check the status of your Fluo instance in YARN.
-When you have data in your fluo instance, you can view it using the command `fluo scan`. Pipe the
+## View Fluo application logs
+
+Fluo application logs are viewable within YARN using the methods below:
+
+1. View logs using the web interface of the the YARN resource manager
+(`http://<resource manager>:8088/cluster`). First, click on the application ID (i.e `application_*`)
+of your Fluo application and then click on the latest attempt ID (i.e `appattempt_*`). You should
+see a list of containers. There should be a container for the application master (typically
+container 1), a Fluo oracle (typically container 2), and Fluo workers (containers 3+). You can view
+the log files produced by a container by clicking on its 'logs' link. Logs from Fluo observers will
+be in the `worker_*.log` file for each of your worker containers.
+
+2. View logs on disk in the directory specified by the YARN property `yarn.nodemanager.log-dirs` in
+your YARN configuration `yarn-site.xml` (see [yarn-default.xml] for more info about this property).
+If you are running Fluo on single machine, this method works well. If you are running Fluo on a
+cluster, your containers and their logs will be spread across the cluster.
+
+When running Fluo in YARN, Fluo must use [logback] for logging (due to a hard requirement by Twill).
+Logback is configured using `/path/to/fluo/apps/<app_name>/conf/logback.xml`. You should review this
+configuration but the root logger is configured by default to print any message that is debug level
+or higher.
+
+In addition the `*.log` files, Fluo oracle and worker containers will have `stdout` and `stderr`
+files. These files may have helpful error messages. Especially, if a process failed to start
+or calls were made to `System.out` or `System.err` in your application.
+
+## View Fluo application data
+
+When you have data in your Fluo application, you can view it using the command `fluo scan`. Pipe the
output to `less` using the command `fluo scan | less` if you want to page through the data.
## Stop your Fluo application
@@ -221,3 +248,5 @@
[fluo.properties]: ../modules/distribution/src/main/config/fluo.properties
[fluo-env.sh]: ../modules/distribution/src/main/config/fluo-env.sh
[lib/ahz/pom.xml]: ../modules/distribution/src/main/lib/ahz/pom.xml
+[yarn-default.xml]: https://hadoop.apache.org/docs/r2.7.0/hadoop-yarn/hadoop-yarn-common/yarn-default.xml
+[logback]: http://logback.qos.ch/
diff --git a/modules/api/src/main/java/org/apache/fluo/api/data/Bytes.java b/modules/api/src/main/java/org/apache/fluo/api/data/Bytes.java
index 2902772..89b111b 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/data/Bytes.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/data/Bytes.java
@@ -29,6 +29,7 @@
import java.util.Objects;
import com.google.common.base.Preconditions;
+import com.google.common.primitives.UnsignedBytes;
/**
* Represents bytes in Fluo. Bytes is an immutable wrapper around a byte array. Bytes always copies
@@ -155,7 +156,7 @@
}
public void writeTo(OutputStream out) throws IOException {
- // since Bytes is immutable, its important the we do not let the internal byte array escape
+ // since Bytes is immutable, its important that we do not let the internal byte array escape
if (length <= 32) {
int end = offset + length;
for (int i = offset; i < end; i++) {
@@ -177,17 +178,23 @@
*/
@Override
public final int compareTo(Bytes other) {
- int minLen = Math.min(this.length(), other.length());
+ if (this == other) {
+ return 0;
+ } else if (this.length == this.data.length && other.length == other.data.length) {
+ return UnsignedBytes.lexicographicalComparator().compare(this.data, other.data);
+ } else {
+ int minLen = Math.min(this.length(), other.length());
- for (int i = 0; i < minLen; i++) {
- int a = (this.byteAt(i) & 0xff);
- int b = (other.byteAt(i) & 0xff);
+ for (int i = 0; i < minLen; i++) {
+ int a = (this.byteAt(i) & 0xff);
+ int b = (other.byteAt(i) & 0xff);
- if (a != b) {
- return a - b;
+ if (a != b) {
+ return a - b;
+ }
}
+ return this.length() - other.length();
}
- return this.length() - other.length();
}
/**
diff --git a/modules/api/src/test/java/org/apache/fluo/api/data/BytesTest.java b/modules/api/src/test/java/org/apache/fluo/api/data/BytesTest.java
index 641c9bf..720a1a3 100644
--- a/modules/api/src/test/java/org/apache/fluo/api/data/BytesTest.java
+++ b/modules/api/src/test/java/org/apache/fluo/api/data/BytesTest.java
@@ -86,10 +86,37 @@
Assert.assertEquals(-1, b1.compareTo(b2));
Assert.assertEquals(1, b2.compareTo(b1));
Assert.assertEquals(0, b1.compareTo(b3));
+ Assert.assertEquals(0, b1.compareTo(b1));
Assert.assertEquals(1, b1.compareTo(Bytes.EMPTY));
}
@Test
+ public void testCompareSubsequence() {
+ Bytes b1 = Bytes.of("abcd");
+ Bytes b2 = b1.subSequence(0, 3);
+ Bytes b3 = Bytes.of("abc");
+ Bytes b4 = Bytes.of("~abcde");
+ Bytes b5 = b4.subSequence(1, 4);
+ Bytes b6 = b4.subSequence(1, 5);
+
+ for (Bytes ba : Arrays.asList(b2, b3, b5, b1.subSequence(0, 3))) {
+ for (Bytes bb : Arrays.asList(b2, b3, b5)) {
+ Assert.assertEquals(0, ba.compareTo(bb));
+ }
+ }
+
+ Assert.assertEquals(1, b1.compareTo(b2));
+ Assert.assertEquals(-1, b2.compareTo(b1));
+
+ for (Bytes less : Arrays.asList(b2, b3, b5)) {
+ for (Bytes greater : Arrays.asList(b1, b4, b6)) {
+ Assert.assertTrue(less.compareTo(greater) < 0);
+ Assert.assertTrue(greater.compareTo(less) > 0);
+ }
+ }
+ }
+
+ @Test
public void testToByteBuffer() {
Bytes b1 = Bytes.of("fluofluo");
ByteBuffer buffer = b1.toByteBuffer();
